RabbitMQ发布确认模式
| 阿里云国内75折 回扣 微信号：monov8 |
| 阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6 |
目录
一、发布确认原理
生产者将信道设置成 confirm 模式一旦信道进入 confirm 模式所有在该信道上面发布的
消息都将会被指派一个唯一的 ID
(从 1 开始)一旦消息被投递到所有匹配的队列之后broker
就会发送一个确认给生产者(包含消息的唯一 ID)这就使得生产者知道消息已经正确到达目的队
列了如果消息和队列是可持久化的那么确认消息会在将消息写入磁盘之后发出broker 回传
给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号此外 broker 也可以设置
basic.ack 的 multiple 域表示到这个序列号之前的所有消息都已经得到了处理。
confirm 模式最大的好处在于他是异步的一旦发布一条消息生产者应用程序就可以在等信
道返回确认的同时继续发送下一条消息当消息最终得到确认之后生产者应用便可以通过回调
方法来处理该确认消息如果 RabbitMQ 因为自身内部错误导致消息丢失就会发送一条 nack 消
息生产者应用程序同样可以在回调方法中处理该 nack 消息。
二、发布确认的策略
一开启发布确认的方法
Channel channel = connection.createChannel();
channel.confirmSelect();二单个确认模式
这是一种简单的确认方式它是一种同步确认发布
的方式也就是发布一个消息之后只有它
被确认发布后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认
的时候才返回如果在指定时间范围内这个消息没有被确认那么它将抛出异常。
这种确认方式有一个最大的缺点就是:发布速度特别的慢
因为如果没有确认发布的消息就会
阻塞所有后续消息的发布这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某
些应用程序来说这可能已经足够了。
public static void publishMessageIndividually() throws Exception {
Channel channel = RabbitMqUtils.getChannel();
channel.confirmSelect();
String queue_name = UUID.randomUUID().toString();
channel.queueDeclare(queue_name, false, false, false, null);
long begin = System.currentTimeMillis();
for (Integer i = 0; i < MESSAGE_COUNT; i++) {
String message = "消息" + i;
channel.basicPublish("", queue_name,null, message.getBytes());
channel.waitForConfirms();
}
long end = System.currentTimeMillis();
System.out.println("发送1000条消息成功耗时为" + (end - begin) + "ms");
}三批量确认模式
上面那种方式非常慢与单个等待确认消息相比先发布一批消息然后一起确认可以极大地
提高吞吐量当然这种方式的缺点就是:当发生故障导致发布出现问题时不知道是哪个消息出现
问题了我们必须将整个批处理保存在内存中以记录重要的信息而后重新发布消息。当然这种
方案仍然是同步的也一样阻塞消息的发布。
public static void publishMessageBatch() throws Exception {
Channel channel = RabbitMqUtils.getChannel();
channel.confirmSelect();
String queue_name = UUID.randomUUID().toString();
channel.queueDeclare(queue_name, false, false, false, null);
long begin = System.currentTimeMillis();
for (Integer i = 0; i < MESSAGE_COUNT; i++) {
if(i % 100 == 0) channel.waitForConfirms();
String message = "消息" + i;
channel.basicPublish("", queue_name,null, message.getBytes());
}
long end = System.currentTimeMillis();
System.out.println("发送1000条消息成功耗时为" + (end - begin) + "ms");
}四异步确认模式
异步确认虽然编程逻辑比上两个要复杂但是性价比最高无论是可靠性还是效率都没得说 他是利用回调函数来达到消息可靠性传递的这个中间件也是通过函数回调来保证是否投递成功。
下面就让我们来详细讲解异步确认是怎么实现的。
public static void publishMessageAsync() throws Exception {
Channel channel = RabbitMqUtils.getChannel();
channel.confirmSelect();
String queue_name = UUID.randomUUID().toString();
channel.queueDeclare(queue_name, false, false, false, null);
ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
// 异步监听broker传递过来的消息确认回调通知
// 确认回调消息
ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
if(multiple) {
// 找出该序号前面所有的消息进行清空
ConcurrentNavigableMap<Long, String> confirmed =
outstandingConfirms.headMap(deliveryTag, true);
//清除该部分未确认消息
confirmed.clear();
} else {
// 不是批量的话就清除单条消息
outstandingConfirms.remove(deliveryTag);
}
System.out.println(deliveryTag + "消息发送成功");
};
// 失败回调消息
ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
// 获取序列号并输出
String message = outstandingConfirms.get(deliveryTag);
System.out.println("发布的消息"+message+"未被确认序列号"+deliveryTag);
};
// 设置确认监听器两个一个监听成功的一个监听失败的
channel.addConfirmListener(ackCallback, nackCallback);
long begin = System.currentTimeMillis();
for (Integer i = 0; i < MESSAGE_COUNT; i++) {
String message = "消息" + i;
channel.basicPublish("", queue_name,null, message.getBytes());
// 获取序列号和消息内容并且把消息放入队列
outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
}
long end = System.currentTimeMillis();
System.out.println("发送1000条消息成功耗时为" + (end - begin) + "ms");
}五如何处理异步未确认消息
最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列
比如说用 ConcurrentLinkedQueue 这个队列在 confirm callbacks 与发布线程之间进行消息的传
递。
六总结
单独发布消息同步等待确认简单但吞吐量非常有限。批量发布消息批量同步等待确认简单合理的吞吐量一旦出现问题但很难推断出是哪条消息出现了问题。异步处理最佳性能和资源使用在出现错误的情况下可以很好地控制但是实现起来稍微难些
public static void main(String[] args) throws Exception {
// publishMessageIndividually(); // 同步确认发送消息, 耗时为769ms
// publishMessageBatch(); // 批量确认发送消息,耗时为84ms
publishMessageAsync(); // 异步确认发送消息, 耗时为43ms
}