如果由于网络缘故原由涌现故障,生产者生产的未到达 Broker 或者 Broker 的被虚假消费,而它们又不知道,就会产生很严重的问题,如重复消费等。
01 RabbitMQ切实其实认流程从图中可以看出:
确认机制分为生产者确认和消费者确认

重点在于生产者重写下面两个方法
rabbitMQTemplate.setConfirmCallbackrabbitMQTemplate.setReturnCallback1.开启生产者确认
spring: rabbitmq: host: localhost port: 5672 virtual-host: / username: root password: root # 开启两个模式的生产者确认 publisher-confirm-type: simple publisher-returns: true
2.声明交流机、行列步队,绑定交流机和行列步队
@Configurationpublic class RabbitMQConfig { private static final String SB_TOPIC_EXCHANGE="sb_topic_exchange"; private static final String SB_TOPIC_QUEUE="sb_topic_queue1"; // 注入交流机 topic类型 @Bean("topicExchange") public Exchange topicExchange(){ return ExchangeBuilder.topicExchange(SB_TOPIC_EXCHANGE).durable(true) .autoDelete().build(); } // 声明行列步队 @Bean public Queue queue1(){ return QueueBuilder.durable(SB_TOPIC_QUEUE).build(); } // 绑定行列步队和交流机 @Bean public Binding exchangQueue(@Qualifier("queue1") Queue queue, @Qualifier("topicExchange") Exchange exchange){ return BindingBuilder.bind(queue).to(exchange).with("user.#").noargs(); }
3.创建消费者
@Component@RabbitListener(queues = "sb_topic_queue1")public class Consumer { @RabbitHandler public void testPublishConfirm(String msg) { System.out.println("收到的信息:"+msg); }}
4.创建生产者
创建生产者发送到行列步队,仿照两种非常情形
@SpringBootTestclass RabiitmqSpringbootApplicationTests { @Autowired RabbitTemplate template; @Test void testConfirmTrue() { // 设置confirm回调函数 template.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean b, java.lang.String s) { if (b) System.out.println("发送成功"); else System.out.println("发送失落败"); } }); // 仿照生产者发送信息--正常情形 template.convertAndSend("sb_topic_exchange","user.info","日志级别:info;日志模块:user;日志信息:"); } @Test void testConfirmFalse() { // 设置confirm回调函数 template.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean b, java.lang.String s) { if (b) System.out.println("发送成功"); else System.out.println("发送失落败"); } }); // 仿照生产者发送信息 // 不存在的交流机--非常情形 template.convertAndSend("sb_topic_exchange_noexist","user.info","日志级别:info;日志模块:user;日志信息:"); } @Test void testReturnFalse() { // 设置return回调函数 template.setReturnCallback(new RabbitTemplate.ReturnCallback() { @Override public void returnedMessage(Message message, int i, java.lang.String s, java.lang.String s1, java.lang.String s2) { System.out.println(message.toString()); System.out.println(s+""); } }); template.setMandatory(true); // 仿照生产者发送信息 // 精确的交流机 缺点的routekey -- 非常情形 template.convertAndSend("sb_topic_exchange","noexist.user.info","日志级别:info;日志模块:user;日志信息:"); }
4.2 消费者确认
重点在于消费者的下面两个方法
channel.basicAck 消费者签收channel.basicNAck 消费者谢绝签收1.开启消费者确认模式
spring: rabbitmq: host: localhost port: 5672 virtual-host: / username: root password: root# 设置消费端手动签收 listener: direct: acknowledge-mode: manual simple: acknowledge-mode: manual
2.创建消费者
/ 注入消费者--手动签到 /@Component@RabbitListener(queues = "sb_topic_queue1")public class Consumer2 { @RabbitHandler public void testComsumer(String msg, Channel channel, Message message) throws InterruptedException, IOException { // 消费端设置手动签收代码 try { System.out.println(msg); // 正常签收,mq收到此被正常签收后即可从行列步队中删除vi信息 // 是哟了那个channel的方法 // 第一个参数是deliverytag 标识哪条信息 第二个参数是是否批量签收 // int i=2/0; 仿照非常 channel.basicAck(message.getMessageProperties().getDeliveryTag(),true); System.out.println("消费者签收了该信息,做事器你可以删了"); }catch (Exception e){ // 非常谢绝签收,让mq重发此信息 System.out.println("该信息丢了,给我重发"); channel.basicNack(message.getMessageProperties().getDeliveryTag(),true,true); // 该信息丢了,但是不须要你重发 // channel.basicNack(message.getMessageProperties().getDeliveryTag(),true,false); } }}
3.创建生产者
@SpringBootTestclass RabiitmqSpringbootApplicationTests { @Autowired RabbitTemplate template; @Test void testConsumerAck() { template.convertAndSend("sb_topic_exchange","noexist.user.info","日志级别:info;日志模块:user;日志信息:"); }}
原文链接:https://www.cnblogs.com/sang-bit/p/14793341.html