MQ(message queue),从字面意思上看,本质是个队列,FIFO 先入先出,只不过队列中存放的内容是message 而已,还是一种跨进程的通信机制,用于上下游传递消息。在互联网架构中,MQ 是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用了 MQ 之后,消息发送上游只需要依赖 MQ,不用依赖其他服务。
RabbitMQ 是一个消息中间件:它接受并转发消息。你可以把它当做一个快递站点,当你要发送一个包裹时,你把你的包裹放到快递站,快递员最终会把你的快递送到件人那里,按照这种逻辑RabbitMQ 是一个快递站,一个快递员帮你传递快件。RabbitMQ 与快递站的主要区别在于,它不处理快件而是接收,存储和转发消息数据。
Broker:接收和分发消息的应用,RabbitMQ Server 就是 Message Broker
Virtual host:出于多租户和安全因素设计的,把 AMQP 的基本组件划分到一个虚拟的分组中,类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时,可以划分出多个 vhost,每个用户在自己的 vhost 创建 exchange/queue 等
Connection:publisher/consumer 和 broker 之间的 TCP 连接
Channel:如果每一次访问 RabbitMQ 都建立一个 Connection,在消息量大的时候建立 TCP
Connection 的开销将是巨大的,效率也较低。Channel 是在 connection 内部建立的逻辑连接,如果应用程序支持多线程,通常每个 thread 创建单独的 channel 进行通讯,AMQP method 包含了 channel id 帮助客户端和 message broker 识别 channel,所以 channel 之间是完全隔离的。Channel 作为轻量级的Connection 极大减少了操作系统建立 TCP connection 的开销
Exchange:message 到达 broker 的第一站,根据分发规则,匹配查询表中的 routing key,分发消息到 queue 中去。常用的类型有:direct (point-to-point), topic (publish-subscribe) and fanout (multicast)
Queue:消息最终被送到这里等待 consumer 取走
Binding:exchange 和 queue 之间的虚拟连接,binding 中可以包含 routing key,Binding 信息被保存到 exchange 中的查询表中,用于 message 的分发依据
<dependencies> <!--rabbitmq 依赖客户端--> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.8.0</version> </dependency> <!--操作文件流的一个依赖--> <dependency> <groupId>commons-io</groupId> <artifactId>commons-io</artifactId> <version>2.6</version> </dependency>
public class Product { public static final String QUEUE_NAME="hollow"; //发消息 public static void main(String[] args) throws Exception { //创建一个连接工厂 ConnectionFactory factory = new ConnectionFactory(); //工厂ip连接rabbit MQ factory.setHost("127.0.0.1"); //用户名 factory.setUsername("guest"); //密码 factory.setPassword("guest"); //创建连接 Connection connection = factory.newConnection(); //获取信道 Channel channel = connection.createChannel(); /** * 生成一个队列 * 1. 队列名称 * 2. 队列里面的消息是否持久化(磁盘)默认情况消息存储在内存中 * 3. 该队列是否只提供一个消费者进行消费 是否进行消息共享 true 多个消费者进行消费 false 一个消费者进行消费 * 4. 是否自动删除 最后一个消费者断开连接以后该队列是否自动删除 true为自动删除 false为不自动删除 * 5. 其他参数 */ channel.queueDeclare(QUEUE_NAME,false,false,false,null); //发送消息 String message="hollow word"; /** * 发送一个消息 * 1.发送到那个交换机 * 2.路由的 key 是哪个 * 3.其他的参数信息 * 4.发送消息的消息体 */ channel.basicPublish("",QUEUE_NAME,null,message.getBytes()); System.out.println("消息发送完毕"); } }
public class Customer { //队列的名称 接收队列中的消息 public static final String QUEUE_NAME="hollow"; public static void main(String[] args) throws Exception { //创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setUsername("guest"); factory.setPassword("guest"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); //声明接收消息 DeliverCallback deliverCallback=(consumerTag,message)->{ System.out.println(message); String s = new String(message.getBody()); System.out.println(s); }; //取消消息是的回调 CancelCallback cancelCallback=consumerTag->{ System.out.println("消息消费被中断"); }; /** * 消费者消费消息 * 1.消费那个消息队列 * 2.消费成功之后是否自动应答 true代表自动应答 false打表手动应答 * 3.消费者为成功消费的回调 * 4.消费者为取消消费的回调 */ channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback); } }
工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务,而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时,这些工作线程将一起处理这些任务。
在这个案例中我们会启动两个工作线程,一个消息发送线程,们两个工作线程接收消息
创建一个RabbitMqUtils用来获取信道
//此类为连接工厂创建信道的工具类 public class RabbitMqUtils { public static Channel getChannel() throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("127.0.0.1"); factory.setUsername("guest"); factory.setPassword("guest"); Connection connection = factory.newConnection(); com.rabbitmq.client.Channel channel = connection.createChannel(); return channel; } }
public class Work01 { //队列的名称 接收队列中的消息 public static final String QUEUE_NAME="hollow"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明接收消息 DeliverCallback deliverCallback=(consumerTag, message)->{ // System.out.println(message); String s = new String(message.getBody()); System.out.println("接收到的消息为"+s); }; //取消消息是的回调 CancelCallback cancelCallback= consumerTag->{ System.out.println(consumerTag+"消息消费被中断"); }; /** * 消费者消费消息 * 1.消费那个消息队列 * 2.消费成功之后是否自动应答 true代表自动应答 false打表手动应答 * 3.消费者为成功消费的回调 * 4.消费者为取消消费的回调 */ System.out.println("C1等待接收消息..."); channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback); } }
在工作空间选择Modify options下的Allow multiple instances 即可以实现多个线程
public class Task01 { public static final String QUEUE_NAME="hollow"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); /** * 生成一个队列 * 1. 队列名称 * 2. 队列里面的消息是否持久化(磁盘)默认情况消息存储在内存中 * 3. 该队列是否只提供一个消费者进行消费 是否进行消息共享 true 多个消费者进行消费 false 一个消费者进行消费 * 4. 是否自动删除 最后一个消费者断开连接以后该队列是否自动删除 true为自动删除 false为不自动删除 * 5. 其他参数 */ channel.queueDeclare(QUEUE_NAME,false,false,false,null); System.out.println("请输入您想发送的消息"); Scanner scanner = new Scanner(System.in); while (scanner.hasNext()) { String message = scanner.next(); /** * 发送一个消息 * 1.发送到那个交换机 * 2.路由的 key 是哪个 * 3.其他的参数信息 * 4.发送消息的消息体 */ channel.basicPublish("",QUEUE_NAME,null,message.getBytes()); System.out.println("消息发送完毕"+message); } } }
消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了,会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息,便立即将该消息标记为删除。在这种情况下,突然有个消费者挂掉了,我们将丢失正在处理的消息。以及后续发送给该消费这的消息,因为它无法接收到。
为了保证消息在发送过程中不丢失,rabbitmq 引入消息应答机制,消息应答就是:消费者在接收到消息并且处理该消息之后,告诉 rabbitmq 它已经处理了,rabbitmq 可以把该消息删除了。
消息发送后立即被认为已经传送成功,这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前,消费者那边出现连接或者 channel 关闭,那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息,没有对传递的消息数量进行限制,当然这样有可能使得消费者这边由于接收太多还来不及处理的消息,导致这些消息的积压,最终使得内存耗尽,最终这些消费者线程被操作系统杀死,所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。
手动应答的好处是可以批量应答并且减少网络拥堵
multiple 的 true 和 false 代表不同意思
如果消费者由于某些原因失去连接(其通道已关闭,连接已关闭或 TCP 连接丢失),导致消息未发送 ACK 确认,RabbitMQ 将了解到消息未完全处理,并将对其重新排队。如果此时其他消费者可以处理,它将很快将其重新分发给另一个消费者。这样,即使某个消费者偶尔死亡,也可以确保不会丢失任何消息。
默认消息采用的是自动应答,所以我们要想实现消息消费过程中不丢失,需要把自动应答改为手动应答
task.java
public class Task { //队列名称 private static final String TASK_QUEUE_NAME="ack_queue"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); //声明队列 channel.queueDeclare(TASK_QUEUE_NAME,false,false,false,null); Scanner scanner = new Scanner(System.in); while (scanner.hasNext()) { String s = scanner.next(); channel.basicPublish("",TASK_QUEUE_NAME,null,s.getBytes(StandardCharsets.UTF_8)); System.out.println("生产者发出消息"+s); } } }
work01.java
public class Work01 { //队列名称 private static final String TASK_QUEUE_NAME="ack_queue"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); System.out.println("C1等待接收消息,处理时间较短"); DeliverCallback deliverCallback=(consumerTag,message)->{ //睡眠1秒 SleepUtils.sleep(1); System.out.println("收到的消息"+new String(message.getBody(),"utf-8")); //手动应答 /** * 1. x消息的标记 tag * 2. 是否批量应答 false 不批量 */ channel.basicAck(message.getEnvelope().getDeliveryTag(),false); }; //采取手动应答 channel.basicConsume(TASK_QUEUE_NAME,false,deliverCallback,(consumerTag->{ System.out.println(consumerTag+"消费者取消消费接口回调逻辑"); })); } }
work02.java
public class Work02 { //队列名称 private static final String TASK_QUEUE_NAME="ack_queue"; public static void main(String[] args) throws Exception { Channel channel = RabbitMqUtils.getChannel(); System.out.println("C2等待接收消息,处理时间较长"); DeliverCallback deliverCallback=(consumerTag,message)->{ //睡眠1秒 SleepUtils.sleep(10); System.out.println("收到的消息"+new String(message.getBody(),"utf-8")); //手动应答 /** * 1. x消息的标记 tag * 2. 是否批量应答 false 不批量 */ channel.basicAck(message.getEnvelope().getDeliveryTag(),false); channel.basicNack(message.getEnvelope().getDeliveryTag(),false,false); }; //采取手动应答 channel.basicConsume(TASK_QUEUE_NAME,false,deliverCallback,(consumerTag->{ System.out.println(consumerTag+"消费者取消消费接口回调逻辑"); })); } }