消息服务器RabbitMQ及其五种消息模型
文章目录
1 消息队列(MQ)是什么
消息队列,即MQ,Message Queue,是一种应用程序对应用程序的通信方法。应用程序通过读取和写出队列中的消息来进行通信,而无需专用连接来链接。
消息队列是典型的:生产者、消费者模型。生产者不断向消息队列中生产消息,消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的,而且只关心消息的发送和接收,没有业务逻辑的侵入,这样就实现了生产者和消费者的解耦。
1.1 实现MQ的两种方式——AMQP/JMS
MQ是消息通信的模型,并发具体实现。现在实现MQ的有两种主流方式:AMQP、JMS
两者间的区别和联系:
- JMS是定义了统一的接口,来对消息操作进行统一;AMQP是通过规定协议来统一数据交互的格式
- JMS限定了必须使用Java语言;AMQP只是协议,不规定实现方式,因此是跨语言的。
- JMS规定了两种消息模型;而AMQP的消息模型更加丰富
常见的MQ产品:
- ActiveMQ:基于JMS
- RabbitMQ:基于AMQP协议,erlang语言(一种通用的面向并发的编程语言)开发,稳定性好
- RocketMQ:基于JMS,阿里巴巴产品,目前交由Apache基金会
- Kafka:分布式消息系统,高吞吐量
1.2 RabbitMQ
RabbitMQ是基于AMQP的一款消息管理系统。一个消息的代理者(Message Broker):它接收消息并且传递消息。
你可以认为它是一个邮局:当你投递邮件到一个邮箱,你很肯定邮递员会终究会将邮件递交给你的收件人。与此类似,RabbitMQ 可以是一个邮箱、邮局、同时还有邮递员。
不同之处在于:RabbitMQ不是传递纸质邮件,而是二进制的数据
官方教程:http://www.rabbitmq.com/getstarted.html
2 五种消息模型
2.1 基本消息模型
生产者和消费者有我们自己来写,本质上RabbitMQ就是上图红色的那部分,存储消息的队列。
在上图的模型中,有以下概念:
- P:生产者,也就是要发送消息的程序
- C:消费者:消息的接受者,会一直等待消息到来。
- queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息。
demo:
连接工具类:
public class ConnectionUtil { /** * 建立与RabbitMQ的连接 * @return * @throws Exception */ public static Connection getConnection() throws Exception { //定义连接工厂 ConnectionFactory factory = new ConnectionFactory(); //设置服务地址 factory.setHost("192.168.56.101"); //端口 factory.setPort(5672); //设置账号信息,用户名、密码、vhost factory.setVirtualHost("/leyou"); factory.setUsername("leyou"); factory.setPassword("leyou"); // 通过工程获取连接 Connection connection = factory.newConnection(); return connection; } }
生产者发送消息:
public class Send { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 从连接中创建通道,使用通道才能完成消息相关的操作 Channel channel = connection.createChannel(); // 声明(创建)队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 消息内容 String message = "Hello World!"; // 向指定的队列中发送消息 channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); //关闭通道和连接 channel.close(); connection.close(); } }
消费者获取消息
public class Recv { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 创建通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [x] received : " + msg + "!"); } }; // 监听队列,第二个参数:是否自动进行消息确认。 channel.basicConsume(QUEUE_NAME, true, consumer); } }
2.1.1 如何避免消息的丢失
当生产者发送出去消息,消费者接收到消息,但是刚接受消息还没有来得及处理就宕机了,但是此时队列中的消息已经被删除了,这显然造成了消息的丢失,如何避免消息的丢失呢?
这就要通过 消息确认机制(Acknowlege) 来实现了。当消费者获取消息后,会向RabbitMQ发送回执ACK,告知消息已经被接收。不过这种回执ACK分两种情况:
- 自动ACK:消息一旦被接收,消费者自动发送ACK
- 手动ACK:消息接收后,不会发送ACK,需要手动调用
大家觉得哪种更好呢?
这需要看消息的重要性:
- 如果消息不太重要,丢失也没有影响,那么自动ACK会比较方便
- 如果消息非常重要,不容丢失。那么最好在消费完成后手动ACK,否则接收消息后就自动ACK,RabbitMQ就会把消息从队列中删除。如果此时消费者宕机,那么消息就丢失了。
我们之前的测试都是自动ACK的,如果要手动ACK,需要改动我们的代码:
// 监听队列,第二个参数false,手动进行ACK channel.basicConsume(QUEUE_NAME, false, consumer);
如果第二个参数为true,则会自动进行ACK;如果为false,则需要手动ACK。
2.2 work消息模型
2.2.1 说明
在刚才的基本模型中,一个生产者,一个消费者,生产的消息直接被消费者消费。比较简单。
Work queues,也被称为(Task queues),任务模型。
当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。比如电商中的短信服务,当用户登陆时候需要短信验证,但是短信发送是非常耗时的,这时候短信发送就会阻塞登录,甚至无法登陆,因此通常的做法就是将短信微服务独立成一个单独的微服务,变同步为异步,这样就可以登录时候异步调用短信微服务往队列里面发送一条消息,但是如果因为某些原因,短信一直发送不出去,用户一直登陆发送短信请求,消息就会堆积越来越多,无法及时处理。此时就可以使用work 模型:让多个消费者绑定到一个队列,共同消费队列中的消息。队列中的消息一旦消费,就会消失,因此任务是不会被重复执行的。
角色:
- P:生产者:任务的发布者
- C1:消费者,领取任务并且完成任务,假设完成速度较慢
- C2:消费者2:领取任务并完成任务,假设完成速度快
生产者与案例1中的几乎一样:
public class Send { private final static String QUEUE_NAME = "test_work_queue"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 循环发布任务 for (int i = 0; i < 50; i++) { // 消息内容 String message = "task .. " + i; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); Thread.sleep(i * 2); } // 关闭通道和连接 channel.close(); connection.close(); } }
消费者1:
消费者2:
与消费者1基本类似,就是没有设置消费耗时时间。这里是模拟有些消费者快,有些比较慢。
不过这里我们是循环发送50条消息。接下来,两个消费者一同启动,然后发送50条消息。结果是两个消费者各自消费了25条消息,而且各不相同,这就实现了任务的分发。
2.2.2 消息堆积越来越多?——能者多劳
刚才的实现有问题吗?
- 消费者1比消费者2的效率要低,一次任务的耗时较长
- 然而两人最终消费的消息数量是一样的
- 消费者2大量时间处于空闲状态,消费者1一直忙碌
现在的状态属于是把任务平均分配,正确的做法应该是消费越快的人,消费的越多。
怎么实现呢?
我们可以修改设置,让消费者同一时间只接收一条消息,这样处理完成之前,就不会接收更多消息,就可以让处理快的人,接收更多消息 :
2.3 三种订阅模型
前面2个案例中,只有3个角色:
- P:生产者,也就是要发送消息的程序
- C:消费者:消息的接受者,会一直等待消息到来。
- queue:消息队列,图中红色部分。类似一个邮箱,可以缓存消息;生产者向其中投递消息,消费者从其中取出消息。
而在订阅模型中,多了一个exchange角色,而且过程略有变化:
- P:生产者,也就是要发送消息的程序,但是不再发送到队列中,而是发给X(交换机)
- C:消费者,消息的接受者,会一直等待消息到来。
- Queue:消息队列,接收消息、缓存消息。
- Exchange:交换机,图中的X。一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。Exchange有以下3种类型: Fanout:广播,将消息交给所有绑定到交换机的队列
- Direct:定向,把消息交给符合指定routing key 的队列
- Topic:通配符,把消息交给符合routing pattern(路由模式) 的队列
Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!
2.3.1 订阅模型-Fanout
Fanout,也称为广播。在广播模式下,消息发送流程是这样的:
- 1) 可以有多个消费者
- 2) 每个消费者有自己的queue(队列)
- 3) 每个队列都要绑定到Exchange(交换机)
- 4) 生产者发送的消息,只能发送到交换机,交换机来决定要发给哪个队列,生产者无法决定。
- 5) 交换机把消息发送给绑定过的所有队列
- 6) 队列的消费者都能拿到消息。实现一条消息被多个消费者消费
生产者:
两个变化:
- 1) 声明Exchange,不再声明Queue
- 2) 发送消息到Exchange,不再发送到Queue
public class Send { private final static String EXCHANGE_NAME = "fanout_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明exchange,指定类型为fanout channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); // 消息内容 String message = "Hello everyone"; // 发布消息到Exchange channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes()); System.out.println(" [生产者] Sent '" + message + "'"); channel.close(); connection.close(); } }
消费者1
public class Recv { private final static String QUEUE_NAME = "fanout_exchange_queue_1"; private final static String EXCHANGE_NAME = "fanout_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者1] received : " + msg + "!"); } }; // 监听队列,自动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }
要注意代码中:队列需要和交换机绑定
消费者2
public class Recv2 { private final static String QUEUE_NAME = "fanout_exchange_queue_2"; private final static String EXCHANGE_NAME = "fanout_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者2] received : " + msg + "!"); } }; // 监听队列,手动返回完成 channel.basicConsume(QUEUE_NAME, true, consumer); } }
2.3.2 订阅模型-Direct
在Fanout模式中,一条消息,会被所有订阅的队列都消费。但是,在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下:
-
队列与交换机的绑定,不能是任意绑定了,而是要指定一个
RoutingKey
(路由key) -
消息的发送方在 向 Exchange发送消息时,也必须指定消息的
RoutingKey
。 -
Exchange不再把消息交给每一个绑定的队列,而是根据消息的
Routing Key
进行判断,只有队列的Routingkey
与消息的Routing key
完全一致,才会接收到消息
图解: -
P:生产者,向Exchange发送消息,发送消息时,会指定一个routing key。
-
X:Exchange(交换机),接收生产者的消息,然后把消息递交给 与routing key完全匹配的队列
-
C1:消费者,其所在队列指定了需要routing key 为 error 的消息
-
C2:消费者,其所在队列指定了需要routing key 为 info、error、warning 的消息
生产者
此处我们模拟商品的增删改,发送消息的RoutingKey分别是:insert、update、delete
public class Send { private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明exchange,指定类型为direct channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 消息内容 String message = "商品新增了, id = 1001"; // 发送消息,并且指定routing key 为:insert ,代表新增商品 channel.basicPublish(EXCHANGE_NAME, "insert", null, message.getBytes()); System.out.println(" [商品服务:] Sent '" + message + "'"); channel.close(); connection.close(); } }
消费者1
我们此处假设消费者1只接收两种类型的消息:更新商品和删除商品。
public class Recv { private final static String QUEUE_NAME = "direct_exchange_queue_1"; private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机,同时指定需要订阅的routing key。假设此处需要update和delete消息 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者1] received : " + msg + "!"); } }; // 监听队列,自动ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }
消费者2
我们此处假设消费者2接收所有类型的消息:新增商品,更新商品和删除商品。
public class Recv2 { private final static String QUEUE_NAME = "direct_exchange_queue_2"; private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机,同时指定需要订阅的routing key。订阅 insert、update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者2] received : " + msg + "!"); } }; // 监听队列,自动ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }
2.3.4 订阅模型-Topic
说明
Topic类型的
Exchange与
Direct相比,都是可以根据
RoutingKey把消息路由到不同的队列。只不过
Topic类型
Exchange可以让队列在绑定
Routing key的时候使用通配符!
Routingkey一般都是有一个或多个单词组成,多个单词之间以”.”分割,例如:
item.insert
通配符规则:
#:匹配一个或多个词
*:匹配不多不少恰好1个词
举例:
audit.#:能够匹配
audit.irs.corporate或者
audit.irs
audit.*:只能匹配
audit.irs
生产者
使用topic类型的Exchange,发送消息的routing key有3种:
item.isnert、
item.update、
item.delete:
public class Send { private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明exchange,指定类型为topic channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 消息内容 String message = "新增商品 : id = 1001"; // 发送消息,并且指定routing key 为:insert ,代表新增商品 channel.basicPublish(EXCHANGE_NAME, "item.insert", null, message.getBytes()); System.out.println(" [商品服务:] Sent '" + message + "'"); channel.close(); connection.close(); } }
消费者1
我们此处假设消费者1只接收两种类型的消息:更新商品和删除商品
public class Recv { private final static String QUEUE_NAME = "topic_exchange_queue_1"; private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机,同时指定需要订阅的routing key。需要 update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.delete"); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者1] received : " + msg + "!"); } }; // 监听队列,自动ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }
消费者2
我们此处假设消费者2接收所有类型的消息:新增商品,更新商品和删除商品。
/** * 消费者2 */ public class Recv2 { private final static String QUEUE_NAME = "topic_exchange_queue_2"; private final static String EXCHANGE_NAME = "topic_exchange_test"; public static void main(String[] argv) throws Exception { // 获取到连接 Connection connection = ConnectionUtil.getConnection(); // 获取通道 Channel channel = connection.createChannel(); // 声明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 绑定队列到交换机,同时指定需要订阅的routing key。订阅 insert、update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "item.*"); // 定义队列的消费者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息体 String msg = new String(body); System.out.println(" [消费者2] received : " + msg + "!"); } }; // 监听队列,自动ACK channel.basicConsume(QUEUE_NAME, true, consumer); } }
3 如何避免消息丢失?——持久化
如何避免消息丢失?
1) 消费者的ACK机制。可以防止消费者丢失消息。
2) 但是,如果在消费者消费之前,MQ就宕机了,消息就没了。
是可以将消息进行持久化呢?要将消息持久化,前提是:队列、Exchange都持久化
交换机持久化:
队列持久化:
消息持久化:
解决消息丢失:
- ACK机制(消费者确认)
- 持久化(交换机持久化、队列持久化、消息持久化)
- 发送消息后,将消息持久化到数据库并记录消息状态(可靠消息服务)
- 生产者确认机制(publisher confirm,MQ向生产者发送)
- RabbitMQ五种消息队列学习(三)--Work模式
- RabbitMQ五种消息队列学习(五)--路由模式
- 消息队列RabbitMq的五种形式队列
- RabbitMQ五种消息队列学习(四)--发布订阅模式
- RabbitMQ五种消息队列学习(六)--通配符模式(路由类型:Topic)
- 消息模型及其伪代码
- JMS消息服务器(二)——点对点消息传送模型
- RabbitMQ消息模型概览(简明教程)
- 如何保证RabbitMQ的消息不丢失及其背后的原理
- JMS消息服务器(二)——点对点消息传送模型
- 【Android】消息系统模型及其Handler、Looper
- web优化必须了解的原理之I/o的五种模型和web的三种工作模式
- Atitit 五种IO模型attilax总结 blocking和non-blocking synchronous IO和asynchronous I
- RabbitMQ消息队列(五):Routing 消息路由
- yii配合rabbitmq邮件消息列队
- rabbitmq 重复ACK导致消息丢失
- arm学习笔记一(arm概述及其基本编程模型)
- RabbitMQ(二):理解消息通信RabbitMQ
- RabbitMQ 消息确认与公平调度消费者
- Windows Socket五种I/O模型——代码全攻略(转载(主要是方便我自己看!^_^~))