在使用 RabbitMQ 的時候,作為訊息發送者希望杜絕任何訊息遺失或投遞失敗場景。 RabbitMQ 為我們提供了兩種方式用來控制訊息的投遞可靠性模式。
confirm 確認模式
return 退回模式
rabbitmq整個訊息投遞的路徑為:
producer—>rabbitmq broker—>exchange—>queue—>consumer
#訊息從producer到ex則會傳回一個confir#Callback
確認模式
訊息從producer 到exchange 則會傳回一個confirmCallback
以spring整合rabbitmq為例,修改rabbitmq設定檔,在connectionFactory中加入publisher-confirms屬性並設定值為true
<!-- * 确认模式: * 步骤: * 1. 确认模式开启:ConnectionFactory中开启publisher-confirms="true" --> <!-- 定义rabbitmq connectionFactory --> <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" virtual-host="${rabbitmq.virtual-host}" publisher-confirms="true"/>
/* * 确认模式: * 步骤: * 2. 在rabbitTemplate定义ConfirmCallBack回调函数 */ @Test public void queueTest(){ rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { /** * * @param correlationData 相关配置信息 * @param ack exchange交换机 是否成功收到了消息。true 成功,false代表失败 * @param cause 失败原因 */ System.out.println("confirm方法被执行了...."); if (ack) { //接收成功 System.out.println("接收成功消息" + cause); } else { //接收失败 System.out.println("接收失败消息" + cause); //做一些处理,让消息再次发送。 } } }); //路由键与队列同名 rabbitTemplate.convertAndSend("spring_queue", "message confirm...."); }
退回模式
訊息從exchange–>queue 投遞失敗則會傳回一個returnCallback
1.開啟回退模式:publisher-returns=「true」
<!-- 定义rabbitmq connectionFactory --> <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" virtual-host="${rabbitmq.virtual-host}" publisher-returns="true"/>
2.設定Exchange處理訊息失敗的模式:setMandatory,然後設定ReturnCallBack
@Test public void queueTest(){ //1.设置交换机处理失败消息的模式 rabbitTemplate.setMandatory(true); //2.设置ReturnCallBack rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { /** * @param message 消息对象 * @param replyCode 错误码 * @param replyText 错误信息 * @param exchange 交换机 * @param routingKey 路由键 */ @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { System.out.println("return 执行了...."); System.out.println(message); System.out.println(replyCode); System.out.println(replyText); System.out.println(exchange); System.out.println(routingKey); //处理 } }); //手动添加错误路由模拟错误发生 rabbitTemplate.convertAndSend("spring_topic_exchange", "return123", "return message..."); }
這裡只有發生錯誤才會回傳訊息,因此手動加上一個錯誤,給發送訊息新增路由值return123,實際上並沒有這個路由,運行返回訊息如下。
Consumer Ack
ack指Acknowledge,確認。表示消費端收到訊息後的確認方式。
其中自動確認是指,當訊息一旦被Consumer接收到,則自動確認收到,並將對應message 從RabbitMQ 的訊息快取中移除。但是在實際業務處理中,很可能訊息會接收到,業務處理出現異常,那麼該訊息就會遺失。如果設定了手動確認方式,則需要在業務處理成功後,呼叫channel.basicAck(),手動簽收,如果出現異常,則呼叫channel.basicNack()方法,讓其自動重新發送訊息。
還是以spring整合rabbitmq為例,rabbitmq設定檔中設定確認方式
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual"> .....
監聽類別程式碼如下:
public class AckListener implements ChannelAwareMessageListener { @Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //1.接收转换消息 System.out.println(new String(message.getBody())); //2. 处理业务逻辑 System.out.println("处理业务逻辑..."); int i = 3/0;//出现错误 // 3. 手动签收 channel.basicAck(deliveryTag,true); } catch (Exception e) { //e.printStackTrace(); //4.拒绝签收 /* *第三个参数:requeue:重回队列。如果设置为true,则消息重新回到queue,broker会 *重新发送该消息给消费端 */ channel.basicNack(deliveryTag,true,true); //channel.basicReject(deliveryTag,true); } } }
因為出現異常呼叫channel.basicNack()方法,讓其自動重新發送訊息,所以無限循環輸出內容
消費端限流
當我們的Rabbitmq 伺服器積壓了有萬條未處理的訊息時,我們隨便打開一個消費者客戶端,會出現這樣情況: 巨量的訊息瞬間全部推送過來,但是我們單一客戶端無法同時處理這麼多資料!當資料量特別大的時候,我們對生產端限流肯定是不科學的,因為有時候並發量就是特別大,有時候並發量又特別少,我們無法約束生產端,這是使用者的行為。所以我們應該對消費端限流,rabbitmq提供了一種qos(服務品質保證)功能,即在非自動確認訊息的前提下,如果一定數目的訊息(給channel或consume設定Qos值)未被確認前,不進行消費新消息。
1.確保ack機制為手動確認
2.listener-container配置屬性perfetch = 1,表示消費端每次從mq拉去一條訊息來消費,直到手動確認消費完畢後,才會繼續拉去下一則訊息。
<rabbit:listener-container connection-factory="connectionFactory" auto-declare="true" acknowledge="manual" prefetch="1"> <rabbit:listener ref="topicListenerACK" queue-names="spring_topic_queue_well2"/> </rabbit:listener-container>
生產者,發送五個訊息
@Test public void topicTest(){ /** * 参数1:交换机名称 * 参数2:路由键名 * 参数3:发送的消息内容 */ for (int i=0;i<5;i++){ rabbitTemplate.convertAndSend("spring_topic_exchange", "xzk.a", "发送到spring_topic_exchange交换机xzk.cn的消息"+i); } } }
生產者註解掉channel.basicAck(deliveryTag,true)即不確認收到訊息
public class AckListener implements ChannelAwareMessageListener { @Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //1.接收转换消息 System.out.println(new String(message.getBody())); //2. 处理业务逻辑 System.out.println("处理业务逻辑..."); // 3. 手动签收 //channel.basicAck(deliveryTag,true); } catch (Exception e) { //e.printStackTrace(); //4.拒绝签收 /* *第三个参数:requeue:重回队列。如果设置为true,则消息重新回到queue,broker会 *重新发送该消息给消费端 */ channel.basicNack(deliveryTag,true,true); } } }
此時啟動消費者再運行生產者之後,發現消費者發送了五條訊息,實際上生產者只接受到了一條訊息,達到限流作用
###觀察rabbitmq控制台,發現有1條unack訊息。 4條ready訊息,還沒到達consumer。和我們設定的prefetchCount=1限流情況相符。 ###把channel.basicAck(deliveryTag,true)的注释取消掉,即可以自动确认收到消息,重新运行消费者,接收到了另外的四条消息
Time To Live,消息过期时间设置
设置交换机,队列以及队列过期时间为10000ms
<!--ttl--> <rabbit:queue name="test_queue_ttl" id="test_queue_ttl"> <rabbit:queue-arguments> <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/> </rabbit:queue-arguments> </rabbit:queue> <rabbit:topic-exchange name="test_exchange_ttl"> <rabbit:bindings> <rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/> </rabbit:bindings> </rabbit:topic-exchange>
生产者发送10条消息
@Test public void testTtl() { for (int i = 0; i < 10; i++) { rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl..."); }
十秒钟后,过期消息消失
设置交换机和队列
<rabbit:queue name="test_queue_ttl" id="test_queue_ttl"/> <rabbit:topic-exchange name="test_exchange_ttl"> <rabbit:bindings> <rabbit:binding pattern="ttl.#" queue="test_queue_ttl"/> </rabbit:bindings> </rabbit:topic-exchange>
生产者发送特定过期消息,用到了MessagePostProcessor这个api
@Test public void testTtl() { MessagePostProcessor messagePostProcessor = new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { //1.设置message信息 message.getMessageProperties().setExpiration("5000");//消息的过期时间 //2.返回该消息 return message; } }; //消息单独过期 rabbitTemplate.convertAndSend("test_exchange_ttl","ttl.hehe","message ttl...",messagePostProcessor); }
5s之后
注:
1.如果同时设置队列过期和消息过期,系统会根据哪个过期的时间短而选用哪儿个。
2.设置单独消息过期时,如果该消息不为第一个接受的消息,则不过期。
死信队列,英文缩写:DLX 。Dead Letter Exchange(死信交换机),当消息成为Deadmessage后,可以被重新发送到另一个交换机,这个交换机就是DLX。
消息成为死信的三种情况:
队列消息长度到达限制;
消费者拒接消费消息,basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false;
原队列存在消息过期设置,消息到达超时时间未被消费;
队列绑定死信交换机:
给队列设置参数: x-dead-letter-exchange 和 x-dead-letter-routing-key
实现
1.声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)
<rabbit:queue name="test_queue_dlx" id="test_queue_dlx"> <!--正常队列绑定死信交换机--> <rabbit:queue-arguments> <!--x-dead-letter-exchange:死信交换机名称--> <entry key="x-dead-letter-exchange" value="exchange_dlx" /> <!--3.2 x-dead-letter-routing-key:发送给死信交换机的routingkey--> <entry key="x-dead-letter-routing-key" value="dlx.hehe" /> <!--4.1 设置队列的过期时间 ttl--> <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/> <!--4.2 设置队列的长度限制 max-length --> <entry key="x-max-length" value="10" value-type="java.lang.Integer" /> </rabbit:queue-arguments> </rabbit:queue> <rabbit:topic-exchange name="test_exchange_dlx"> <rabbit:bindings> <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"> </rabbit:binding> </rabbit:bindings> </rabbit:topic-exchange>
2.声明死信队列(queue_dlx)和死信交换机(exchange_dlx)
<rabbit:queue name="queue_dlx" id="queue_dlx"></rabbit:queue> <rabbit:topic-exchange name="exchange_dlx"> <rabbit:bindings> <rabbit:binding pattern="dlx.#" queue="queue_dlx"></rabbit:binding> </rabbit:bindings> </rabbit:topic-exchange>
3.生产端测试
/** * 发送测试死信消息: * 1. 过期时间 * 2. 长度限制 * 3. 消息拒收 */ @Test public void testDlx(){ //1. 测试过期时间,死信消息 rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?"); //2. 测试长度限制后,消息死信 /* for (int i = 0; i < 20; i++) { rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?"); }*/ //3. 测试消息拒收 //rabbitTemplate.convertAndSend("test_exchange_dlx","test.dlx.haha","我是一条消息,我会死吗?"); }
4.消费端监听
public class DlxListener implements ChannelAwareMessageListener { @Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //1.接收转换消息 System.out.println(new String(message.getBody())); //2. 处理业务逻辑 System.out.println("处理业务逻辑..."); int i = 3/0;//出现错误 //3. 手动签收 channel.basicAck(deliveryTag,true); } catch (Exception e) { //e.printStackTrace(); System.out.println("出现异常,拒绝接受"); //4.拒绝签收,不重回队列 requeue=false channel.basicNack(deliveryTag,true,false); } } }
<rabbit:listener ref="dlxListener" queue-names="test_queue_dlx"> </rabbit:listener>
延迟队列,即消息进入队列后不会立即被消费,只有到达指定时间后,才会被消费。c
需求:
1.下单后,30分钟未支付,取消订单,回滚库存。
2.新用户注册成功7天后,发送短信问候。
实现方式:
定时器
延迟队列
定时器的实现方式不够优雅,我们采取延迟队列的方式
不过很可惜,在RabbitMQ中并未提供延迟队列功能。
但是可以使用:TTL+死信队列 组合实现延迟队列的效果。
配置
<!-- 延迟队列: 1. 定义正常交换机(order_exchange)和队列(order_queue) 2. 定义死信交换机(order_exchange_dlx)和队列(order_queue_dlx) 3. 绑定,设置正常队列过期时间为30分钟 --> <!-- 定义正常交换机(order_exchange)和队列(order_queue)--> <rabbit:queue id="order_queue" name="order_queue"> <!-- 绑定,设置正常队列过期时间为30分钟--> <rabbit:queue-arguments> <entry key="x-dead-letter-exchange" value="order_exchange_dlx" /> <entry key="x-dead-letter-routing-key" value="dlx.order.cancel" /> <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer"/> </rabbit:queue-arguments> </rabbit:queue> <rabbit:topic-exchange name="order_exchange"> <rabbit:bindings> <rabbit:binding pattern="order.#" queue="order_queue"></rabbit:binding> </rabbit:bindings> </rabbit:topic-exchange> <!-- 定义死信交换机(order_exchange_dlx)和队列(order_queue_dlx)--> <rabbit:queue id="order_queue_dlx" name="order_queue_dlx"></rabbit:queue> <rabbit:topic-exchange name="order_exchange_dlx"> <rabbit:bindings> <rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx"></rabbit:binding> </rabbit:bindings> </rabbit:topic-exchange>
生产端测试
@Test public void testDelay() throws InterruptedException { //1.发送订单消息。 将来是在订单系统中,下单成功后,发送消息 rabbitTemplate.convertAndSend("order_exchange","order.msg","订单信息:id=1,time=2019年8月17日16:41:47"); /*//2.打印倒计时10秒 for (int i = 10; i > 0 ; i--) { System.out.println(i+"..."); Thread.sleep(1000); }*/ }
消费端监听
public class OrderListener implements ChannelAwareMessageListener { @Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //1.接收转换消息 System.out.println(new String(message.getBody())); //2. 处理业务逻辑 System.out.println("处理业务逻辑..."); System.out.println("根据订单id查询其状态..."); System.out.println("判断状态是否为支付成功"); System.out.println("取消订单,回滚库存...."); //3. 手动签收 channel.basicAck(deliveryTag,true); } catch (Exception e) { //e.printStackTrace(); System.out.println("出现异常,拒绝接受"); //4.拒绝签收,不重回队列 requeue=false channel.basicNack(deliveryTag,true,false); } } }
<rabbit:listener ref="orderListener" queue-names="order_queue_dlx"> </rabbit:listener>
以上是Java RabbitMQ高階特性實例分析的詳細內容。更多資訊請關注PHP中文網其他相關文章!