文章来源&＃xff1a;【公众号&＃xff1a;楼仔】
目录
背景
消息队列
RabbitMQ 原理
RabbitMQ 实战
RabbitMQ 进阶
背景
常见的消息队列很多&＃xff0c;主要包括 RabbitMQ、Kafka、RocketMQ 和 ActiveMQ&＃xff0c;相关的选型可以看我之前的系列&＃xff0c;这篇文章只讲 RabbitMQ&＃xff0c;先讲原理&＃xff0c;后搞实战。
文章很长&＃xff0c;如果你能一次性看完&＃xff0c;“大哥&＃xff0c;请收下我的膝盖”&＃xff0c;建议大家先收藏&＃xff0c;啥时需要面试&＃xff0c;或者工作中遇到了&＃xff0c;可以再慢慢看。
不 BB&＃xff0c;直接上思维导图&＃xff1a;
消息队列
消息队列目前主要 2 种模式&＃xff0c;分别为“点对点模式”和“发布/订阅模式”。
一个具体的消息只能由一个消费者消费&＃xff0c;多个生产者可以向同一个消息队列发送消息&＃xff0c;但是一个消息在被一个消息者处理的时候&＃xff0c;这个消息在队列上会被锁住或者被移除并且其他消费者无法处理该消息。
需要额外注意的是&＃xff0c;如果消费者处理一个消息失败了&＃xff0c;消息系统一般会把这个消息放回队列&＃xff0c;这样其他消费者可以继续处理。
单个消息可以被多个订阅者并发的获取和处理。
一般来说&＃xff0c;订阅有两种类型&＃xff1a;
临时&＃xff08;ephemeral&＃xff09;订阅&＃xff1a;这种订阅只有在消费者启动并且运行的时候才存在。一旦消费者退出&＃xff0c;相应的订阅以及尚未处理的消息就会丢失。
持久&＃xff08;durable&＃xff09;订阅&＃xff1a;这种订阅会一直存在&＃xff0c;除非主动去删除。消费者退出后&＃xff0c;消息系统会继续维护该订阅&＃xff0c;并且后续消息可以被继续处理。
对消息队列进行技术选型时&＃xff0c;需要通过以下指标衡量你所选择的消息队列&＃xff0c;是否可以满足你的需求&＃xff1a;
消息顺序&＃xff1a;发送到队列的消息&＃xff0c;消费时是否可以保证消费的顺序&＃xff0c;比如 A 先下单&＃xff0c;B 后下单&＃xff0c;应该是 A 先去扣库存&＃xff0c;B 再去扣&＃xff0c;顺序不能反。
消息路由&＃xff1a;根据路由规则&＃xff0c;只订阅匹配路由规则的消息&＃xff0c;比如有 A/B 两者规则的消息&＃xff0c;消费者可以只订阅 A 消息&＃xff0c;B 消息不会消费。
消息可靠性&＃xff1a;是否会存在丢消息的情况&＃xff0c;比如有 A/B 两个消息&＃xff0c;最后只有 B 消息能消费&＃xff0c;A 消息丢失。
消息时序&＃xff1a;主要包括“消息存活时间”和“延迟/预定的消息”&＃xff0c;“消息存活时间”表示生产者可以对消息设置 TTL&＃xff0c;如果超过该 TTL&＃xff0c;消息会自动消失&＃xff1b;“延迟/预定的消息”指的是可以延迟或者预订消费消息&＃xff0c;比如延时 5 分钟&＃xff0c;那么消息会 5 分钟后才能让消费者消费&＃xff0c;时间未到的话&＃xff0c;是不能消费的。
消息留存&＃xff1a;消息消费成功后&＃xff0c;是否还会继续保留在消息队列。
容错性&＃xff1a;当一条消息消费失败后&＃xff0c;是否有一些机制&＃xff0c;保证这条消息是一种能成功&＃xff0c;比如异步第三方退款消息&＃xff0c;需要保证这条消息消费掉&＃xff0c;才能确定给用户退款成功&＃xff0c;所以必须保证这条消息消费成功的准确性。
伸缩&＃xff1a;当消息队列性能有问题&＃xff0c;比如消费太慢&＃xff0c;是否可以快速支持库容&＃xff1b;当消费队列过多&＃xff0c;浪费系统资源&＃xff0c;是否可以支持缩容。
吞吐量&＃xff1a;支持的最高并发数。
RabbitMQ 原理初探
RabbitMQ 2007 年发布&＃xff0c;是使用 Erlang 语言开发的开源消息队列系统&＃xff0c;基于 AMQP 协议来实现。
提到 RabbitMQ&＃xff0c;就不得不提 AMQP 协议。AMQP 协议是具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议&＃xff0c;是应用层协议的一个开放标准&＃xff0c;为面向消息的中间件设计。
先了解一下 AMQP 协议中间的几个重要概念&＃xff1a;
Server&＃xff1a;接收客户端的连接&＃xff0c;实现 AMQP 实体服务。
Connection&＃xff1a;连接&＃xff0c;应用程序与 Server 的网络连接&＃xff0c;TCP 连接。
Channel&＃xff1a;信道&＃xff0c;消息读写等操作在信道中进行。客户端可以建立多个信道&＃xff0c;每个信道代表一个会话任务。
Message&＃xff1a;消息&＃xff0c;应用程序和服务器之间传送的数据&＃xff0c;消息可以非常简单&＃xff0c;也可以很复杂。由 Properties 和 Body 组成。Properties 为外包装&＃xff0c;可以对消息进行修饰&＃xff0c;比如消息的优先级、延迟等高级特性&＃xff1b;Body 就是消息体内容。
Virtual Host&＃xff1a;虚拟主机&＃xff0c;用于逻辑隔离。一个虚拟主机里面可以有若干个 Exchange 和 Queue&＃xff0c;同一个虚拟主机里面不能有相同名称的 Exchange 或 Queue。
Exchange&＃xff1a;交换器&＃xff0c;接收消息&＃xff0c;按照路由规则将消息路由到一个或者多个队列。如果路由不到&＃xff0c;或者返回给生产者&＃xff0c;或者直接丢弃。RabbitMQ 常用的交换器常用类型有 direct、topic、fanout、headers 四种&＃xff0c;后面详细介绍。
Binding&＃xff1a;绑定&＃xff0c;交换器和消息队列之间的虚拟连接&＃xff0c;绑定中可以包含一个或者多个 RoutingKey。
RoutingKey&＃xff1a;路由键&＃xff0c;生产者将消息发送给交换器的时候&＃xff0c;会发送一个 RoutingKey&＃xff0c;用来指定路由规则&＃xff0c;这样交换器就知道把消息发送到哪个队列。路由键通常为一个“.”分割的字符串&＃xff0c;例如“com.rabbitmq”。
Queue&＃xff1a;消息队列&＃xff0c;用来保存消息&＃xff0c;供消费者消费。
AMQP 协议模型由三部分组成&＃xff1a;生产者、消费者和服务端。
执行流程如下&＃xff1a;
生产者是连接到 Server&＃xff0c;建立一个连接&＃xff0c;开启一个信道。
生产者声明交换器和队列&＃xff0c;设置相关属性&＃xff0c;并通过路由键将交换器和队列进行绑定。
消费者也需要进行建立连接&＃xff0c;开启信道等操作&＃xff0c;便于接收消息。
生产者发送消息&＃xff0c;发送到服务端中的虚拟主机。
虚拟主机中的交换器根据路由键选择路由规则&＃xff0c;发送到不同的消息队列中。
订阅了消息队列的消费者就可以获取到消息&＃xff0c;进行消费。
RabbitMQ 常用的交换器类型有 direct、topic、fanout、headers 四种&＃xff1a;
Direct Exchange&＃xff1a;见文知意&＃xff0c;直连交换机意思是此交换机需要绑定一个队列&＃xff0c;要求该消息与一个特定的路由键完全匹配。简单点说就是一对一的&＃xff0c;点对点的发送。
Fanout Exchange&＃xff1a;这种类型的交换机需要将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播&＃xff0c;每台子网内的主机都获得了一份复制的消息。简单点说就是发布订阅。
Topic Exchange&＃xff1a;直接翻译的话叫做主题交换机&＃xff0c;如果从用法上面翻译可能叫通配符交换机会更加贴切。这种交换机是使用通配符去匹配&＃xff0c;路由到对应的队列。
通配符有两种&＃xff1a;"*" 、 "#"&＃xff0c;需要注意的是通配符前面必须要加上"."符号&＃xff1a;
*符号&＃xff1a;有且只匹配一个词。比如 a.*可以匹配到"a.b"、"a.c"&＃xff0c;但是匹配不了"a.b.c"。
#符号&＃xff1a;匹配一个或多个词。比如"rabbit.#"既可以匹配到"rabbit.a.b"、"rabbit.a"&＃xff0c;也可以匹配到"rabbit.a.b.c"。
Headers Exchange&＃xff1a;这种交换机用的相对没这么多。它跟上面三种有点区别&＃xff0c;它的路由不是用routingKey进行路由匹配&＃xff0c;而是在匹配请求头中所带的键值进行路由。创建队列需要设置绑定的头部信息&＃xff0c;有两种模式&＃xff1a;全部匹配和部分匹配。
如上图所示&＃xff0c;交换机会根据生产者发送过来的头部信息携带的键值去匹配队列绑定的键值&＃xff0c;路由到对应的队列。
我们先看几个基本概念&＃xff1a;
broker&＃xff1a;每个节点运行的服务程序&＃xff0c;功能为维护该节点的队列的增删以及转发队列操作请求。
master queue&＃xff1a;每个队列都分为一个主队列和若干个镜像队列。
mirror queue&＃xff1a;镜像队列&＃xff0c;作为 master queue 的备份。在 master queue 所在节点挂掉之后&＃xff0c;系统把 mirror queue 提升为 master queue&＃xff0c;负责处理客户端队列操作请求。注意&＃xff0c;mirror queue 只做镜像&＃xff0c;设计目的不是为了承担客户端读写压力。
集群中有两个节点&＃xff0c;每个节点上有一个 broker&＃xff0c;每个 broker 负责本机上队列的维护&＃xff0c;并且 borker 之间可以互相通信。
集群中有两个队列 A 和 B&＃xff0c;每个队列都分为 master queue 和 mirror queue&＃xff08;备份&＃xff09;。那么队列上的生产消费怎么实现的呢&＃xff1f;
对于消费队列&＃xff0c;如下图有两个 consumer 消费队列 A&＃xff0c;这两个 consumer 连在了集群的不同机器上。
RabbitMQ 集群中的任何一个节点都拥有集群上所有队列的元信息&＃xff0c;所以连接到集群中的任何一个节点都可以&＃xff0c;主要区别在于有的 consumer 连在 master queue 所在节点&＃xff0c;有的连在非 master queue 节点上。
因为 mirror queue 要和 master queue 保持一致&＃xff0c;故需要同步机制&＃xff0c;正因为一致性的限制&＃xff0c;导致所有的读写操作都必须都操作在 master queue 上&＃xff08;想想&＃xff0c;为啥读也要从 master queue 中读&＃xff1f;和数据库读写分离是不一样的&＃xff09;&＃xff0c;然后由 master 节点同步操作到 mirror queue 所在的节点。
即使 consumer 连接到了非 master queue 节点&＃xff0c;该 consumer 的操作也会被路由到 master queue 所在的节点上&＃xff0c;这样才能进行消费。
对于生成队列&＃xff0c;原理和消费一样&＃xff0c;如果连接到非 master queue 节点&＃xff0c;则路由过去。
Time To Live&＃xff0c;也就是生存时间&＃xff0c;是一条消息在队列中的最大存活时间&＃xff0c;单位是毫秒。
下面看看 RabbitMQ 过期时间特性&＃xff1a;
RabbitMQ 可以对消息和队列设置 TTL。
RabbitMQ 支持设置消息的过期时间&＃xff0c;在消息发送的时候可以进行指定&＃xff0c;每条消息的过期时间可以不同。
RabbitMQ 支持设置队列的过期时间&＃xff0c;从消息入队列开始计算&＃xff0c;直到超过了队列的超时时间配置&＃xff0c;那么消息会变成死信&＃xff0c;自动清除。
如果两种方式一起使用&＃xff0c;则过期时间以两者中较小的那个数值为准。
当然也可以不设置 TTL&＃xff0c;不设置表示消息不会过期&＃xff1b;如果设置为 0&＃xff0c;则表示除非此时可以直接将消息投递到消费者&＃xff0c;否则该消息将被立即丢弃。
为了保证消息从队列可靠地到达消费者&＃xff0c;RabbitMQ 提供了消息确认机制。
消费者订阅队列的时候&＃xff0c;可以指定 autoAck 参数&＃xff0c;当 autoAck 为 true 的时候&＃xff0c;RabbitMQ 采用自动确认模式&＃xff0c;RabbitMQ 自动把发送出去的消息设置为确认&＃xff0c;然后从内存或者硬盘中删除&＃xff0c;而不管消费者是否真正消费到了这些消息。
当 autoAck 为 false 的时候&＃xff0c;RabbitMQ 会等待消费者回复的确认信号&＃xff0c;收到确认信号之后才从内存或者磁盘中删除消息。
消息确认机制是 RabbitMQ 消息可靠性投递的基础&＃xff0c;只要设置 autoAck 参数为 false&＃xff0c;消费者就有足够的时间处理消息&＃xff0c;不用担心处理消息的过程中消费者进程挂掉后消息丢失的问题。
消息的可靠性是 RabbitMQ 的一大特色&＃xff0c;那么 RabbitMQ 是如何保证消息可靠性的呢&＃xff1f;
答案就是消息持久化。持久化可以防止在异常情况下丢失数据。RabbitMQ 的持久化分为三个部分&＃xff1a;交换器持久化、队列持久化和消息的持久化。
交换器持久化可以通过在声明队列时将 durable 参数设置为 true。如果交换器不设置持久化&＃xff0c;那么在 RabbitMQ 服务重启之后&＃xff0c;相关的交换器元数据会丢失&＃xff0c;不过消息不会丢失&＃xff0c;只是不能将消息发送到这个交换器了。
队列的持久化能保证其本身的元数据不会因异常情况而丢失&＃xff0c;但是不能保证内部所存储的消息不会丢失。
要确保消息不会丢失&＃xff0c;需要将其设置为持久化。队列的持久化可以通过在声明队列时将 durable 参数设置为 true。
设置了队列和消息的持久化&＃xff0c;当 RabbitMQ 服务重启之后&＃xff0c;消息依然存在。如果只设置队列持久化或者消息持久化&＃xff0c;重启之后消息都会消失。
当然&＃xff0c;也可以将所有的消息都设置为持久化&＃xff0c;但是这样做会影响 RabbitMQ 的性能&＃xff0c;因为磁盘的写入速度比内存的写入要慢得多。
对于可靠性不是那么高的消息可以不采用持久化处理以提高整体的吞吐量。鱼和熊掌不可兼得&＃xff0c;关键在于选择和取舍。在实际中&＃xff0c;需要根据实际情况在可靠性和吞吐量之间做一个权衡。
当消息在一个队列中变成死信之后&＃xff0c;他能被重新发送到另一个交换器中&＃xff0c;这个交换器成为死信交换器&＃xff0c;与该交换器绑定的队列称为死信队列。
消息变成死信有下面几种情况&＃xff1a;
消息被拒绝。
消息过期
队列达到最大长度
DLX 也是一个正常的交换器&＃xff0c;和一般的交换器没有区别&＃xff0c;他能在任何的队列上面被指定&＃xff0c;实际上就是设置某个队列的属性。
当这个队列中有死信的时候&＃xff0c;RabbitMQ 会自动将这个消息重新发送到设置的交换器上&＃xff0c;进而被路由到另一个队列&＃xff0c;我们可以监听这个队列中消息做相应的处理。
死信队列有什么用&＃xff1f;当发生异常的时候&＃xff0c;消息不能够被消费者正常消费&＃xff0c;被加入到了死信队列中。后续的程序可以根据死信队列中的内容分析当时发生的异常&＃xff0c;进而改善和优化系统。
一般的队列&＃xff0c;消息一旦进入队列就会被消费者立即消费。延迟队列就是进入该队列的消息会被消费者延迟消费&＃xff0c;延迟队列中存储的对象是的延迟消息&＃xff0c;“延迟消息”是指当消息被发送以后&＃xff0c;等待特定的时间后&＃xff0c;消费者才能拿到这个消息进行消费。
延迟队列用于需要延迟工作的场景。最常见的使用场景&＃xff1a;淘宝或者天猫我们都使用过&＃xff0c;用户在下单之后通常有 30 分钟的时间进行支付&＃xff0c;如果这 30 分钟之内没有支付成功&＃xff0c;那么订单就会自动取消。
除了延迟消费&＃xff0c;延迟队列的典型应用场景还有延迟重试。比如消费者从队列里面消费消息失败了&＃xff0c;可以延迟一段时间以后进行重试。
这里才是内容的重点&＃xff0c;不仅需要知道 Rabbit 的特性&＃xff0c;还需要知道支持这些特性的原因&＃xff1a;
消息路由&＃xff08;支持&＃xff09;&＃xff1a;RabbitMQ 可以通过不同的交换器支持不同种类的消息路由
消息有序&＃xff08;不支持&＃xff09;&＃xff1a;当消费消息时&＃xff0c;如果消费失败&＃xff0c;消息会被放回队列&＃xff0c;然后重新消费&＃xff0c;这样会导致消息无序
消息时序&＃xff08;非常好&＃xff09;&＃xff1a;通过延时队列&＃xff0c;可以指定消息的延时时间&＃xff0c;过期时间 TTL 等
容错处理&＃xff08;非常好&＃xff09;&＃xff1a;通过交付重试和死信交换器&＃xff08;DLX&＃xff09;来处理消息处理故障
伸缩&＃xff08;一般&＃xff09;&＃xff1a;伸缩其实没有非常智能&＃xff0c;因为即使伸缩了&＃xff0c;master queue 还是只有一个&＃xff0c;负载还是只有这一个 master queue 去抗&＃xff0c;所以我理解 RabbitMQ 的伸缩很弱&＃xff08;个人理解&＃xff09;
持久化&＃xff08;不太好&＃xff09;&＃xff1a;没有消费的消息&＃xff0c;可以支持持久化&＃xff0c;这个是为了保证机器宕机时消息可以恢复&＃xff0c;但是消费过的消息&＃xff0c;就会被马上删除&＃xff0c;因为 RabbitMQ 设计时&＃xff0c;就不是为了去存储历史数据的
消息回溯&＃xff08;不支持&＃xff09;&＃xff1a;因为消息不支持永久保存&＃xff0c;所以自然就不支持回溯
高吞吐&＃xff08;中等&＃xff09;&＃xff1a;因为所有的请求的执行&＃xff0c;最后都是在 master queue&＃xff0c;它的这个设计&＃xff0c;导致单机性能达不到十万级的标准
RabbitMQ 环境搭建
因为我用的是 Mac&＃xff0c;所以直接可以参考官网&＃xff1a;
https://www.rabbitmq.com/install-homebrew.html
需要注意的是&＃xff0c;一定需要先执行&＃xff1a;
brew update
然后再执行&＃xff1a;
brew install rabbitmq
之前没有执行 brew update&＃xff0c;直接执行 brew install rabbitmq 时&＃xff0c;会报各种各样奇怪的错误&＃xff0c;其中“403 Forbidde”居多。
但是在执行“brew install rabbitmq”&＃xff0c;会自动安装其它的程序&＃xff0c;如果你使用源码安装 Rabbitmq&＃xff0c;因为启动该服务依赖 erlang 环境&＃xff0c;所以你还需手动安装 erlang&＃xff0c;但是目前官方已经一键给你搞定&＃xff0c;会自动安装 Rabbitmq 依赖的所有程序&＃xff0c;是不是很棒&＃xff01;
最后执行成功的输出如下&＃xff1a;
启动服务&＃xff1a;
# 启动方式1&＃xff1a;后台启动
brew services start rabbitmq
# 启动方式2&＃xff1a;当前窗口启动
cd /usr/local/Cellar/rabbitmq/3.8.19
rabbitmq-server
在浏览器输入&＃xff1a;
http://localhost:15672/
会出现 RabbitMQ 后台管理界面&＃xff08;用户名和密码都为 guest&＃xff09;&＃xff1a;
通过 brew 安装&＃xff0c;一行命令搞定&＃xff0c;真香&＃xff01;
RabbitMQ 测试
首先得启动 mq&＃xff1a;
## 添加账号
./rabbitmqctl add_user admin admin
## 添加访问权限
./rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
## 设置超级权限
./rabbitmqctl set_user_tags admin administrator
因为代码中引入了 java 8 的特性&＃xff0c;pom 引入依赖&＃xff1a;
开始写代码&＃xff1a;
public class RabbitMqTest {//消息队列名称private final static String QUEUE_NAME &＃61; "hello";&＃64;Testpublic void send() throws java.io.IOException, TimeoutException {//创建连接工程ConnectionFactory factory &＃61; new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setUsername("admin");factory.setPassword("admin");//创建连接Connection connection &＃61; factory.newConnection();//创建消息通道Channel channel &＃61; connection.createChannel();//生成一个消息队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);for (int i &＃61; 0; i < 10; i&＃43;&＃43;) {String message &＃61; "Hello World RabbitMQ count: " &＃43; i;//发布消息&＃xff0c;第一个参数表示路由&＃xff08;Exchange名称&＃xff09;&＃xff0c;为""则表示使用默认消息路由channel.basicPublish("", QUEUE_NAME, null, message.getBytes());System.out.println(" [x] Sent &＃39;" &＃43; message &＃43; "&＃39;");}//关闭消息通道和连接channel.close();connection.close();}&＃64;Testpublic void consumer() throws java.io.IOException, TimeoutException {//创建连接工厂ConnectionFactory factory &＃61; new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setUsername("admin");factory.setPassword("admin");//创建连接Connection connection &＃61; factory.newConnection();//创建消息信道final Channel channel &＃61; connection.createChannel();//消息队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);System.out.println("[*] Waiting for message. To exist press CTRL&＃43;C");DeliverCallback deliverCallback &＃61; (consumerTag, delivery) -> {String message &＃61; new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received &＃39;" &＃43; message &＃43; "&＃39;");};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});}
}
执行 send() 后控制台输出&＃xff1a;
[x] Sent &＃39;Hello World RabbitMQ count: 0&＃39;
[x] Sent &＃39;Hello World RabbitMQ count: 1&＃39;
[x] Sent &＃39;Hello World RabbitMQ count: 2&＃39;
[x] Sent &＃39;Hello World RabbitMQ count: 3&＃39;
[x] Sent &＃39;Hello World RabbitMQ count: 4&＃39;
[x] Sent &＃39;Hello World RabbitMQ count: 5&＃39;
[x] Sent &＃39;Hello World RabbitMQ count: 6&＃39;
[x] Sent &＃39;Hello World RabbitMQ count: 7&＃39;
[x] Sent &＃39;Hello World RabbitMQ count: 8&＃39;
[x] Sent &＃39;Hello World RabbitMQ count: 9&＃39;
执行consumer()后&＃xff1a;
示例中的代码讲解&＃xff0c;可以直接参考官网&＃xff1a;https://www.rabbitmq.com/tutorials/tutorial-one-java.html
基本使用姿势
封装工厂类&＃xff1a;
public class RabbitUtil {public static ConnectionFactory getConnectionFactory() {//创建连接工程&＃xff0c;下面给出的是默认的caseConnectionFactory factory &＃61; new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setUsername("admin");factory.setPassword("admin");factory.setVirtualHost("/");return factory;}
}
封装生成者&＃xff1a;
public class MsgProducer {public static void publishMsg(String exchange, BuiltinExchangeType exchangeType, String toutingKey, String message) throws IOException, TimeoutException {ConnectionFactory factory &＃61; RabbitUtil.getConnectionFactory();//创建连接Connection connection &＃61; factory.newConnection();//创建消息通道Channel channel &＃61; connection.createChannel();// 声明exchange中的消息为可持久化&＃xff0c;不自动删除channel.exchangeDeclare(exchange, exchangeType, true, false, null);// 发布消息channel.basicPublish(exchange, toutingKey, null, message.getBytes());System.out.println("Sent &＃39;" &＃43; message &＃43; "&＃39;");channel.close();connection.close();}
}
封装消费者&＃xff1a;
public class MsgConsumer {public static void consumerMsg(String exchange, String queue, String routingKey)throws IOException, TimeoutException {ConnectionFactory factory &＃61; RabbitUtil.getConnectionFactory();//创建连接Connection connection &＃61; factory.newConnection();//创建消息信道final Channel channel &＃61; connection.createChannel();//消息队列channel.queueDeclare(queue, true, false, false, null);//绑定队列到交换机channel.queueBind(queue, exchange, routingKey);System.out.println("[*] Waiting for message. To exist press CTRL&＃43;C");Consumer consumer &＃61; new DefaultConsumer(channel) {&＃64;Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {String message &＃61; new String(body, "UTF-8");try {System.out.println(" [x] Received &＃39;" &＃43; message);} finally {System.out.println(" [x] Done");channel.basicAck(envelope.getDeliveryTag(), false);}}};// 取消自动ackchannel.basicConsume(queue, false, consumer);}
}
生产者&＃xff1a;
public class DirectProducer {private static final String EXCHANGE_NAME &＃61; "direct.exchange";public void publishMsg(String routingKey, String msg) {try {MsgProducer.publishMsg(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, routingKey, msg);} catch (Exception e) {e.printStackTrace();}}public static void main(String[] args) throws InterruptedException {DirectProducer directProducer &＃61; new DirectProducer();String[] routingKey &＃61; new String[]{"aaa", "bbb", "ccc"};String msg &＃61; "hello >>> ";for (int i &＃61; 0; i < 10; i&＃43;&＃43;) {directProducer.publishMsg(routingKey[i % 3], msg &＃43; i);}System.out.println("----over-------");Thread.sleep(1000 * 60 * 100);}
}
执行生产者&＃xff0c;往消息队列中放入 10 条消息&＃xff0c;其中 key 分别为“aaa”、“bbb”和“ccc”&＃xff0c;分别放入 qa、qb、qc 三个队列&＃xff1a;
下面是 qa 队列的信息&＃xff1a;
消费者&＃xff1a;
public class DirectConsumer {private static final String exchangeName &＃61; "direct.exchange";public void msgConsumer(String queueName, String routingKey) {try {MsgConsumer.consumerMsg(exchangeName, queueName, routingKey);} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}public static void main(String[] args) throws InterruptedException {DirectConsumer consumer &＃61; new DirectConsumer();String[] routingKey &＃61; new String[]{"aaa", "bbb", "ccc"};String[] queueNames &＃61; new String[]{"qa", "qb", "qc"};for (int i &＃61; 0; i < 3; i&＃43;&＃43;) {consumer.msgConsumer(queueNames[i], routingKey[i]);}Thread.sleep(1000 * 60 * 100);}
}
执行后的输出&＃xff1a;
[*] Waiting for message. To exist press CTRL&＃43;C[x] Received &＃39;hello >>> 0[x] Done[x] Received &＃39;hello >>> 3[x] Done[x] Received &＃39;hello >>> 6[x] Done[x] Received &＃39;hello >>> 9[x] Done
[*] Waiting for message. To exist press CTRL&＃43;C[x] Received &＃39;hello >>> 1[x] Done[x] Received &＃39;hello >>> 4[x] Done[x] Received &＃39;hello >>> 7[x] Done
[*] Waiting for message. To exist press CTRL&＃43;C[x] Received &＃39;hello >>> 2[x] Done[x] Received &＃39;hello >>> 5[x] Done[x] Received &＃39;hello >>> 8[x] Done
可以看到&＃xff0c;分别从 qa、qb、qc 中将不同的 key 的数据消费掉。
我做了个简单的实验&＃xff0c;我把消费者代码修改了一下&＃xff1a;
public static void main(String[] args) throws InterruptedException {DirectConsumer consumer &＃61; new DirectConsumer();String[] routingKey &＃61; new String[]{"aaa", "bbb", "ccc"};String[] queueNames &＃61; new String[]{"qa", "qb", "qc1"}; // 将qc修改为qc1for (int i &＃61; 0; i < 3; i&＃43;&＃43;) {consumer.msgConsumer(queueNames[i], routingKey[i]);}Thread.sleep(1000 * 60 * 100);
}
执行后如下图所示&＃xff1a;
我们可以发现&＃xff0c;多了一个 qc1&＃xff0c;所以可以判断这个界面中的 queues&＃xff0c;是消费者执行时&＃xff0c;会将消费者指定的队列名称和 direct.exchange 绑定&＃xff0c;绑定的依据就是 key。
当我们把队列中的数据全部消费掉&＃xff0c;然后重新执行生成者后&＃xff0c;会发现 qc 和 qc1 中都有 3 条待消费的数据&＃xff0c;因为绑定的 key 都是“ccc”&＃xff0c;所以两者的数据是一样的&＃xff1a;
绑定关系如下&＃xff1a;
注意&＃xff1a;当没有 Queue 绑定到 Exchange 时&＃xff0c;往 Exchange 中写入的消息也不会重新分发到之后绑定的 queue 上。
思考&＃xff1a;不执行消费者&＃xff0c;看不到这个 Queres 中信息&＃xff0c;我其实可以把这个界面理解为消费者信息界面。
不过感觉还是怪怪的&＃xff0c;这个 queues 如果是消费者信息&＃xff0c;就不应该叫 queues&＃xff0c;我理解 queues 应该是 RabbitMQ 中实际存放数据的 queues&＃xff0c;难道是我理解错了&＃xff1f;
生产者封装&＃xff1a;
public class FanoutProducer {private static final String EXCHANGE_NAME &＃61; "fanout.exchange";public void publishMsg(String routingKey, String msg) {try {MsgProducer.publishMsg(EXCHANGE_NAME, BuiltinExchangeType.FANOUT, routingKey, msg);} catch (Exception e) {e.printStackTrace();}}public static void main(String[] args) {FanoutProducer directProducer &＃61; new FanoutProducer();String msg &＃61; "hello >>> ";for (int i &＃61; 0; i < 10; i&＃43;&＃43;) {directProducer.publishMsg("", msg &＃43; i);}}
}
消费者&＃xff1a;
public class FanoutConsumer {private static final String EXCHANGE_NAME &＃61; "fanout.exchange";public void msgConsumer(String queueName, String routingKey) {try {MsgConsumer.consumerMsg(EXCHANGE_NAME, queueName, routingKey);} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}public static void main(String[] args) {FanoutConsumer consumer &＃61; new FanoutConsumer();String[] queueNames &＃61; new String[]{"qa-2", "qb-2", "qc-2"};for (int i &＃61; 0; i < 3; i&＃43;&＃43;) {consumer.msgConsumer(queueNames[i], "");}}
}
执行生成者&＃xff0c;结果如下&＃xff1a;
我们发现&＃xff0c;生产者生产的 10 条数据&＃xff0c;在每个消费者中都可以消费&＃xff0c;这个是和 Direct 不同的地方。
但是使用 Fanout 方式时&＃xff0c;有几个点需要注意一下&＃xff1a;
生产者的 routkey 可以为空&＃xff0c;因为生产者的所有数据&＃xff0c;会下放到每一个队列&＃xff0c;所以不会通过 routkey 去路由。
消费者需要指定 queues&＃xff0c;因为消费者需要绑定到指定的 queues 才能消费。
这幅图就画出了 Fanout 的精髓之处&＃xff0c;exchange 会和所有的 queue 进行绑定&＃xff0c;不区分路由&＃xff0c;消费者需要绑定指定的 queue 才能发起消费。
注意&＃xff1a;往队列塞数据时&＃xff0c;可能通过界面看不到消息个数的增加&＃xff0c;可能是你之前已经开启了消费进程&＃xff0c;导致增加的消息马上被消费了。
上面我们是指定了队列&＃xff0c;这个方式其实很不友好&＃xff0c;比如对于 Fanout&＃xff0c;我其实根本无需关心队列的名字&＃xff0c;如果还指定对应队列进行消费&＃xff0c;感觉这个很冗余&＃xff0c;所以我们这里就采用随机获取队列名字的方式&＃xff0c;下面代码直接 Copy 官网。
生成者封装&＃xff1a;
public static void publishMsgV2(String exchange, BuiltinExchangeType exchangeType, String message) throws IOException, TimeoutException {ConnectionFactory factory &＃61; RabbitUtil.getConnectionFactory();//创建连接Connection connection &＃61; factory.newConnection();//创建消息通道Channel channel &＃61; connection.createChannel();// 声明exchange中的消息channel.exchangeDeclare(exchange, exchangeType);// 发布消息channel.basicPublish(exchange, "", null, message.getBytes("UTF-8"));System.out.println("Sent &＃39;" &＃43; message &＃43; "&＃39;");channel.close();connection.close();
}
消费者封装&＃xff1a;
public static void consumerMsgV2(String exchange) throws IOException, TimeoutException {ConnectionFactory factory &＃61; RabbitUtil.getConnectionFactory();Connection connection &＃61; factory.newConnection();final Channel channel &＃61; connection.createChannel();channel.exchangeDeclare(exchange, "fanout");String queueName &＃61; channel.queueDeclare().getQueue();channel.queueBind(queueName, exchange, "");System.out.println(" [*] Waiting for messages. To exit press CTRL&＃43;C");DeliverCallback deliverCallback &＃61; (consumerTag, delivery) -> {String message &＃61; new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received &＃39;" &＃43; message &＃43; "&＃39;");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
}
生产者&＃xff1a;
public class FanoutProducer {private static final String EXCHANGE_NAME &＃61; "fanout.exchange.v2";public void publishMsg(String msg) {try {MsgProducer.publishMsgV2(EXCHANGE_NAME, BuiltinExchangeType.FANOUT, msg);} catch (Exception e) {e.printStackTrace();}}public static void main(String[] args) {FanoutProducer directProducer &＃61; new FanoutProducer();String msg &＃61; "hello >>> ";for (int i &＃61; 0; i < 10000; i&＃43;&＃43;) {directProducer.publishMsg(msg &＃43; i);}}
}
消费者&＃xff1a;
public class FanoutConsumer {private static final String EXCHANGE_NAME &＃61; "fanout.exchange.v2";public void msgConsumer() {try {MsgConsumer.consumerMsgV2(EXCHANGE_NAME);} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}}public static void main(String[] args) {FanoutConsumer consumer &＃61; new FanoutConsumer();for (int i &＃61; 0; i < 3; i&＃43;&＃43;) {consumer.msgConsumer();}}
}
执行后&＃xff0c;管理界面如下&＃xff1a;
代码详见官网&＃xff1a;
https://www.rabbitmq.com/tutorials/tutorial-five-java.html
更多方式&＃xff0c;请直接查看官网&＃xff1a;
https://www.rabbitmq.com/getstarted.html
RabbitMQ 进阶
在定义 Queue 时&＃xff0c;可以指定这两个参数&＃xff1a;
/*** Declare an exchange.* &＃64;see com.rabbitmq.client.AMQP.Exchange.Declare* &＃64;see com.rabbitmq.client.AMQP.Exchange.DeclareOk* &＃64;param exchange the name of the exchange* &＃64;param type the exchange type* &＃64;param durable true if we are declaring a durable exchange (the exchange will survive a server restart)* &＃64;param autoDelete true if the server should delete the exchange when it is no longer in use* &＃64;param arguments other properties (construction arguments) for the exchange* &＃64;return a declaration-confirm method to indicate the exchange was successfully declared* &＃64;throws java.io.IOException if an error is encountered*/
Exchange.DeclareOk exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete,Map
* Declare a queue
* &＃64;see com.rabbitmq.client.AMQP.Queue.Declare
* &＃64;see com.rabbitmq.client.AMQP.Queue.DeclareOk
* &＃64;param queue the name of the queue
* &＃64;param durable true if we are declaring a durable queue (the queue will survive a server restart)
* &＃64;param exclusive true if we are declaring an exclusive queue (restricted to this connection)
* &＃64;param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
* &＃64;param arguments other properties (construction arguments) for the queue
* &＃64;return a declaration-confirm method to indicate the queue was successfully declared
* &＃64;throws java.io.IOException if an error is encountered
*/
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map
持久化&＃xff0c;保证 RabbitMQ 在退出或者 crash 等异常情况下数据没有丢失&＃xff0c;需要将 queue&＃xff0c;exchange 和 Message 都持久化。
若是将 queue 的持久化标识 durable 设置为 true&＃xff0c;则代表是一个持久的队列&＃xff0c;那么在服务重启之后&＃xff0c;会重新读取之前被持久化的 queue。
虽然队列可以被持久化&＃xff0c;但是里面的消息是否为持久化&＃xff0c;还要看消息的持久化设置。即重启 queue&＃xff0c;但是 queue 里面还没有发出去的消息&＃xff0c;那队列里面还存在该消息么&＃xff1f;这个取决于该消息的设置。
自动删除&＃xff0c;如果该队列没有任何订阅的消费者的话&＃xff0c;该队列会被自动删除。这种队列适用于临时队列。
当一个 Queue 被设置为自动删除时&＃xff0c;当消费者断掉之后&＃xff0c;queue 会被删除&＃xff0c;这个主要针对的是一些不是特别重要的数据&＃xff0c;不希望出现消息积累的情况。
当一个 Queue 已经声明好了之后&＃xff0c;不能更新 durable 或者 autoDelted 值&＃xff1b;当需要修改时&＃xff0c;需要先删除再重新声明
消费的 Queue 声明应该和投递的 Queue 声明的 durable&＃xff0c;autoDelted 属性一致&＃xff0c;否则会报错
对于重要的数据&＃xff0c;一般设置 durable&＃61;true&＃xff0c;autoDeleted&＃61;false
对于设置 autoDeleted&＃61;true 的队列&＃xff0c;当没有消费者之后&＃xff0c;队列会自动被删除
执行一个任务可能需要花费几秒钟&＃xff0c;你可能会担心如果一个消费者在执行任务过程中挂掉了。
一旦 RabbitMQ 将消息分发给了消费者&＃xff0c;就会从内存中删除。在这种情况下&＃xff0c;如果正在执行任务的消费者宕机&＃xff0c;会丢失正在处理的消息和分发给这个消费者但尚未处理的消息。
但是&＃xff0c;我们不想丢失任何任务&＃xff0c;如果有一个消费者挂掉了&＃xff0c;那么我们应该将分发给它的任务交付给另一个消费者去处理。
为了确保消息不会丢失&＃xff0c;RabbitMQ 支持消息应答。消费者发送一个消息应答&＃xff0c;告诉 RabbitMQ 这个消息已经接收并且处理完毕了。RabbitMQ 就可以删除它了。
因此手动 ACK 的常见手段&＃xff1a;
// 接收消息之后&＃xff0c;主动ack/nak
Consumer consumer &＃61; new DefaultConsumer(channel) {&＃64;Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,byte[] body) throws IOException {String message &＃61; new String(body, "UTF-8");try {System.out.println(" [ " &＃43; queue &＃43; " ] Received &＃39;" &＃43; message);channel.basicAck(envelope.getDeliveryTag(), false);} catch (Exception e) {channel.basicNack(envelope.getDeliveryTag(), false, true);}}
};
// 取消自动ack
channel.basicConsume(queue, false, consumer);
------------- END -------------
扫码免费获取600&＃43;页石杉老师原创精品文章汇总PDF
原创技术文章汇总
点个在看你最好看
京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有 
