网站专题策划,中国电信六大外包公司,广州代注册公司哪家好,无极兼职网发布确认模式是避免消息由生产者到RabbitMQ消息丢失的一种手段 发布确认模式 原理说明实现方式开启confirm#xff08;确认#xff09;模式阻塞确认异步确认 总结 原理说明 生产者通过调用channel.confirmSelect方法将信道设置为confirm模式#xff0c;之后RabbitMQ会返回Co… 发布确认模式是避免消息由生产者到RabbitMQ消息丢失的一种手段 发布确认模式 原理说明实现方式开启confirm确认模式阻塞确认异步确认 总结 原理说明 生产者通过调用channel.confirmSelect方法将信道设置为confirm模式之后RabbitMQ会返回Confirm.Select-OK命令表示同意生产者将当前信道设置为confirm模式。 confirm模式下的信道所发送的消息都将被应带ack或者nack一次不会出现一条消息即被ack又被nack的情况并且RabbitMQ也并没有对消息被confirm的快慢做出保证消息被confirm是异步进行。 如上图所示为confirm模式下的消息发送过程其中4和6为异步应答也就是说4过程并不一定在5之前也有可能是在下一条消息发送后才会进行上一条消息的应答。 RabbitMQ 事务和发送确认机制确保的是消息能够正确的发送至RabbitMQ的交换机如果交换机没有匹配的队列那么消息也会被丢失。和事务不同的是发布确认机制是异步进行的因此在性能上发布确认模式将更加优秀需要注意的是事务和确认机制是互斥的不能共存。 事务机制和发布确认机制都存在以下注意点
如果消息需要持久化并且存在队列则在消息入队并且持久化后进行返回事务提交成功或者应答消息。如果消息不需要持久化但是存在队列则在消息入队后返回事务提交成功或者应答消息。如果消息不可路由到队列中则在路由失败后返回事务提交成功或者应答消息。 上文中一直强调的时发布确认针对发布发送到RabbitMQ中的交换机进行保证但消息实际是否能入队发布确认机制并不能提供保证因此还需要和mandatory参数配合使用。
实现方式 RabbitMQ的发布确认机制可以分为三种实现方式阻塞等待确认、批量阻塞等待确认、异步确认。 阻塞等待确认每当消息发送后发送者都阻塞的等待应答消息。这种实现方式将无法体现发布确认模式的异步性能优势。 批量阻塞确认批量阻塞确认类似于阻塞等待确认区别在于批量阻塞确认并不会针对每条消息进行阻塞等待他会针对一些消息进行统一阻塞等待应答消息。这种实现方式将同步和异步结合起来进行使用对应答性能有一定的提升。 异步应答实现一个监听器的方式接收应答消息应答消息的处理逻辑不会影响消息的发送消息的应答和消息发送是异步进行的他们并不直接相互干扰。 上面对三种确认方式进行简单说明下面将分别介绍发布确认机制的实现方式。
开启confirm确认模式 确认模式的开启是针对信道设置的一旦信道进入了confirm模式所有在该信道上面发布的消息都会被指派唯一的IDRabbitMQ也将针对该信道发送的所有消息都进行应答。 RabbitMQ回传给生产者的确认消息中的deliverryTag包含了确认消息的序号但在使用批量阻塞确认方式进行实现的时候该消息序号无意义。开启confirm模式仅需要以下代码进行实现即可
channel.confirmSelect();阻塞确认 阻塞确认的方式依赖于channel.waitForConfirms()方法该方法如下所示 /*** Wait until all messages published since the last call have been* either ackd or nackd by the broker. Note, when called on a* non-Confirm channel, waitForConfirms throws an IllegalStateException.* return whether all the messages were ackd (and none were nackd)* throws java.lang.IllegalStateException*/boolean waitForConfirms() throws InterruptedException;自从上次调用该方法后直到所有发送的消息都被应答后返回所有消息的应答结果如果所有发送的消息应答结果都是成功则返回true一旦存在任何一条消息应答失败则返回false。 根据该方法的描述可知可以通过该方法实现阻塞等待确认和批量阻塞确认两种方案区别仅在于是发送一条消息调用一次该方法还是发送一批消息后调用一次这个方法。 阻塞等待确认的方式如下代码所示
//发送消息
channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
// 如果发送失败则进行该条消息的重新发送
if(!channel.waitForConfirms()){channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
}阻塞批量确认的方式如下代码所示 // 存储未应带消息队列ListString messages new ArrayList();for (int i 1; i 20000 ; i){String msg String.valueOf(i);messages.add(msg);channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());// 每发送十条消息进行一次确认if(i 0 i % 10 0 ){// 如果确认不通过则将消息重新发送if(!channel.waitForConfirms()){for (String e : messages) {channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,e.getBytes());}}else{// 如果确认成功则将这些消息从未应答队列中移除messages.clear();}}}异步确认 客户端Channel提供了addConfirmListener方法该可以添加ConfirmListener这个回调接口该接口包含两个方法handleAck和handleNack分别用来处理饭hi的Ack和Nack这两个方法都将返回一个参数deliveryTag消息的唯一有序序号和一个boolean型参数multiple如果该参数为true表示自该消息之前的所有消息RabbitMQ服务都已经做出了应答。我们可以通过该值实现具体业务的发布确认。
/**
* Implement this interface in order to be notified of Confirm events.
* Acks represent messages handled successfully; Nacks represent
* messages lost by the broker. Note, the lost messages could still
* have been delivered to consumers, but the broker cannot guarantee
* this.
* For a lambda-oriented syntax, use {link ConfirmCallback}.
*/
public interface ConfirmListener {void handleAck(long deliveryTag, boolean multiple)throws IOException;void handleNack(long deliveryTag, boolean multiple)throws IOException;
}异步确认的方式实现起来比较复杂在生产者端需要维护一个消息队列如果消息应答成功则将该消息从队列中移除如果消息应答失败则将该消息再重新发送或进行其他业务处理。该逻辑伪码如下所示 // 存储未确认消息其中key为消息序号value为消息实体HashMapLong,String msgMap new HashMap();channel.addConfirmListener(new ConfirmListener() {Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {msgMap.remove(deliveryTag);}/*** 如果消息应带结果为nack则重新发送该消息* param deliveryTag* param multiple* throws IOException*/Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {String msg msgMap.get(deliveryTag);if(msg ! null){channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());}}});for (int i 1; i 20000 ; i){String msg String.valueOf(i);// 将消息序号和消息存储map中msgMap.put(channel.getNextPublishSeqNo(),msg);channel.basicPublish(EXCHANGE_NAME ,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());}上述代码使用了map存储消息序号和消息实体这种存储方式应该会存在风险由于监听器和消息发送过程是异步进行了因此可能会存在线程安全的问题HashMap是非线程安全的。
总结 发布确认模式是为我们解决消息自生产者发送到RabbitMQ交换机过程中消息丢失的问题的这一场景需求我们也可以通过事务机制实现。发布确认模式和事务机制比较如下表所示
比较事务机制发布确认机制实现方式通过AMQP协议层面实现轻量级实现采用RabbitMQ应答机制命令详解Tx.SelectBasic.PublishTx.CommitCommit.OKBasic.PublishBasic.Ack性能同步性能较慢可异步实现也可同步实现性能快AMQP命令交互少消息到达队列时机事务提交后消息才会进入队列消息入队存在滞后性消息发送后就进入队列发布确认模式不影响消息进入队列时机事务提交成功或消息应答时机消息被交换机处理完成后或消息不可达同事务实现复杂度简单相对复杂适合场景批量发送消息实现批量消息的原子性和一致性确保消息发送到交换机 发布确认模式的具体实现可以划分为三种阻塞等待、批量确认、异步确认这三者的比较如下表所示
比较内容阻塞等待批量等待异步确认性能低中高实现复杂度低中高确认范围每条消息批量消息每条消息是否可以精准确认每条消息是否是 根据上述内容我们在实现避免消息自生产者到交换机丢失的机制时建议使用发布确认模式的异步确认因为异步确认性能最高并且可以准确的得到被应答的消息的序号有助于我们进行后续逻辑处理。
