消息的投递有4个环节, 如下图
.
环节1: 生产者Producer把消息发送给服务器Broker
Producer怎么知道Broker有没有接收到消息
服务端确认-Transaction模式
只要channel.txCommit()方法返回, 服务端就一定接收到了消息
缺点: 同步模式, 等Broker返回成功之后 , Producer才会继续发送下一条消息, 大大降低了效率
服务端确认-Confirm模式
消息发送一条, 确认一条, 一样很慢
下面的代码是批量的发送一批消息, 确认一批消息
只要waitForConfirmOrDie方法没有抛异常, 那么服务器就批量接收到了一批消息, 这个时候, 效率是有提升的
但是批次的数量很有讲究, 如果每批数量少了,那提升的效率很有限. 如果一批的数量过多, 那么只要有一条消息出错, 就会导致这批里的所有消息都要重发. 所以也是不够完美
服务端确认 - 异步确认模式
这才是完美的解决方案
先定义一个线程安全的confirmSet, 每次Producer发消息的时候, 会顺带往这个集合里扔消息id进去
在handleNack方法中, 是Broker未确认的消息, 可以存起来, 在后面重新发送
在handleAck方法中, 是Broker确认的消息, 确认的消息, 就从confirmSet中删除
控制台输出:
.
环节2: 消息到达Exchange之后, Exchange会查找路由规则列表, 将消息投递到一个或多个Queue上
问题: 某条消息发到Exchange之后, 根据路由规则找不到Queue
解决方案: 路由保证
- mandatory = true + ReturnListener
- 指定交换机的备份交换机
s1参数是用来做路由的, 如果根据Exchange的路由规则, 没有对应的Queue, 那么会去看第三个参数mandatory. 如果mandatory是false, 那么消息会被直接丢弃, 如果是true, 那么就会把这条消息转给另一个指定的Exchange, 下图是教你如何指定这个ReturnExchange的
.
在控制台界面里配置Arguments, 添加一行alternate-exchange
.
环节3: 消息在Queue上存储, 还没有Consumer来消费, 会一直存在Queue中
问题: 消息默认存在Broker内存中, Broker重启会导致消息丢失
解决方案是要确保如下几条同时满足
- Exchange要持久化
- Queue要持久化
- 消息本身也要持久化
- Broker集群 ( 如果只有单台Broker, 那么即使是开启了持久化, 一旦这个Broker硬盘挂了, 消息还是会全部丢失 )
环节4:Queue里的消息一条一条投递到Consumer
Queue如何知道Consumer成功消费了消息
消费者确认机制
- Consumer自动ack: 只要接收到消息, 就发送ack给Broker, 所以会导致漏处理 ( 不推荐 )
- Consumer手动ack: 可以在接收到消息, 并且处理完业务逻辑之后, 才发ack给Broker
下面是Consumer端的推荐写法: 开启手工应答, 并根据处理各种不同情况分别做 reject / nack / ack 的响应给Broker
集成Spring的时候, AcknowledgeMode的枚举类型起名容易引起歧义