手机号码网站开发,西安seo网站排名,商城网站前期seo应该怎么做,vi设计RabbitMQ消息投递的路径为#xff1a; 生产者 — 交换机 — 队列 — 消费者 在RabbitMQ工作的过程中#xff0c;每个环节消息都可能传递失败#xff0c;那么RabbitMQ是如何监听消息是否成功投递的呢#xff1f; 确认模式#xff08;confirm#xff09;可以监…RabbitMQ消息投递的路径为 生产者 — 交换机 — 队列 — 消费者 在RabbitMQ工作的过程中每个环节消息都可能传递失败那么RabbitMQ是如何监听消息是否成功投递的呢 确认模式confirm可以监听消息是否从生产者成功传递到交换机。 退回模式return可以监听消息是否从交换机成功传递到队列。 消费者消息确认Consumer Ack可以监听消费者是否成功处理消息。
开启确认模式
publisher-confirm-type: correlated
SpringBootTest
public class ProducerTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testConfirm(){// 定义确认模式的回调方法消息向交换机发送后会调用confirm方法rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*** 被调用的回调方法* param correlationData 相关配置信息* param ack 交换机是否成功收到了消息* param cause 失败原因*/Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {if (ack){System.out.println(confirm接受成功!);}else{System.out.println(confirm接受失败原因为cause);// 做一些处理。}}});rabbitTemplate.convertAndSend(my_topic_exchange,my_routing,send message...);}
}
消息回退
spring:rabbitmq:host: 192.168.0.162port: 5672username: guestpassword: guestvirtual-host: /# 开启确认模式publisher-confirm-type: correlated# 开启回退模式publisher-returns: true
SpringBootTest
public class ProducerTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testReturn(){// 定义退回模式的回调方法。交换机发送到队列失败后才会执行returnedMessage方法rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {/*** param returned 失败后将失败信息封装到参数中*/Overridepublic void returnedMessage(ReturnedMessage returned) {System.out.println(消息对象returned.getMessage());System.out.println(错误码returned.getReplyCode());System.out.println(错误信息returned.getReplyText());System.out.println(交换机returned.getExchange());System.out.println(路由键returned.getRoutingKey());// 处理消息...}});rabbitTemplate.convertAndSend(my_topic_exchange,my_routing1,send message...);}
}
消息确认 自动确认spring.rabbitmq.listener.simple.acknowledge“none” 手动确认spring.rabbitmq.listener.simple.acknowledge“manual” 消费者处理消息时定义手动签收和拒绝签收的情况
Component
public class AckConsumer {RabbitListener(queues my_queue)public void listenMessage(Message message, Channel channel) throws IOException, InterruptedException {// 消息投递序号消息每次投递该值都会1long deliveryTag message.getMessageProperties().getDeliveryTag();try {int i 1/0; //模拟处理消息出现bugSystem.out.println(成功接受到消息:message);// 签收消息/*** 参数1消息投递序号* 参数2是否一次可以签收多条消息*/channel.basicAck(deliveryTag,true);}catch (Exception e){System.out.println(消息消费失败);Thread.sleep(2000);// 拒签消息/*** 参数1消息投递序号* 参数2是否一次可以拒签多条消息* 参数3拒签后消息是否重回队列*/channel.basicNack(deliveryTag,true,true);}}
}
