springBoot集成rabbitmq 之消息确认模式(ACK)

╰半橙微兮° 2024-03-27 16:33 56阅读 0赞

### springBoot集成rabbitmq 之消息确认模式(ACK) ###

#### comsumer服务 ####

配置

server:
      port: 8081
    spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest #用户名
        password: guest  #密码
        virtual-host: /
        publisher-confirms: true  #支持发布确认
        publisher-returns: true   #支持发布返回
        listener:
          simple:
            acknowledge-mode: manual    #acknowledgeMode设置为手动模式(NONE,MANUAL,AUTO)
            concurrency: 1              #当前监听容器数
            max-concurrency: 1          #最大数
            retry:
              enabled: true             #是否支持重试
          direct:
            acknowledge-mode: manual    #acknowledgeMode设置为手动模式

配置类

@Configuration
    public class RabbitAckConfig {
         
        /**
         * Queue 可以有4个参数
         *      1.name          队列名
         *      2.durable       持久化消息队列 ,rabbitmq重启的时候不需要创建新的队列 默认true
         *      3.auto-delete   表示消息队列没有在使用时将被自动删除 默认是false
         *      4.exclusive     表示该消息队列是否只在当前connection生效,默认是false
         */
        @Bean
        public Queue createAckQueue() {
        
            return new Queue("Queue_ACK",true);
        }
    }

接收者A 和 B，B是必定报错的，解释在代码里面。

import com.alibaba.fastjson.JSON;
    import com.lideru.consumer.pojo.User;
    import com.rabbitmq.client.Channel; 
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
     
    @Component
    @RabbitListener(queues = "Queue_ACK")
    public class AckReceiverA {
        
        @RabbitHandler
        public void process(String str,Channel channel, Message message)  {
        
            User user =  JSON.parseObject(str,User.class);
            System.out.println("消费者A : " + user.toString() + new Date());
            try {
        
                //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                System.out.println("消息消费A成功！");
            } catch (Exception e) {
        
                System.out.println("消息消费A失败:"+ e.getMessage() + e);
                //丢弃这条消息
                //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            }
            try {
        
                Thread.sleep(1000);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    
    
    //消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息
    //channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    //ack返回false，并重新回到队列，api里面解释得很清楚
    //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    //拒绝消息
    //channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
    
    }

import com.alibaba.fastjson.JSON;
    import com.lideru.consumer.pojo.User;
    import com.rabbitmq.client.Channel; 
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    import java.util.Date;
     
    @Component
    @RabbitListener(queues = "Queue_ACK")
    public class AckReceiverB {
        
        @Autowired
        private AckSender ackSender;
        @RabbitHandler
        public void process(String str,Channel channel, Message message) throws InterruptedException, IOException {
        
            User user =  JSON.parseObject(str,User.class);
            System.out.println("消费者A : " + user.toString() + new Date());
            try {
        
                int i = 1/0; // 报错，重新发送mq。
                //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                System.out.println("消息消费B成功！");
            } catch (Exception e) {
        
                // 捕获异常，删除mq，重新发送。
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                System.out.println("消息消费B失败:"+ e.getMessage() + e);
                ackSender.send(user);
                //丢弃这条消息
                //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            }
            Thread.sleep(5000);
        }
    
    //消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息
    //channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    //ack返回false，并重新回到队列，api里面解释得很清楚
    //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    //拒绝消息
    //channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
    
    }

发送者

import com.alibaba.fastjson.JSON;
    import com.lideru.consumer.pojo.User;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    import java.util.UUID;
    
    
    @Component
    public class AckSender implements  RabbitTemplate.ConfirmCallback ,RabbitTemplate.ReturnCallback {
        
    
    	@Autowired
    	private RabbitTemplate rabbitTemplate;
    
    	public void send(User user) {
        
    		//设置回调对象
    		rabbitTemplate.setConfirmCallback(this);
    		rabbitTemplate.setReturnCallback(this);
    		//构建回调返回的数据
    		CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    		String content = JSON.toJSONString(user);
    		rabbitTemplate.convertAndSend("Queue_ACK",(Object) content,correlationData);
    		System.out.println("确认发送ok ,"+new Date()+","+content);
    	}
    
    	/**
    	 * 消息回调确认方法
    	 * 如果消息没有到exchange,则confirm回调,ack=false
    	 * 如果消息到达exchange,则confirm回调,ack=true
    	 * @param
    	 */
    	@Override
    	public void confirm(CorrelationData correlationData, boolean isSendSuccess, String s) {
        
    		System.out.println("confirm--message:回调消息ID为: " + correlationData.getId());
    		if (isSendSuccess) {
        
    			System.out.println("confirm--message:消息发送成功");
    		} else {
        
    			System.out.println("确认--消息:消息发送失败" + s);
    		}
    	}
    
    	/**
    	 * exchange到queue成功,则不回调return
    	 * exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
    	 */
    	@Override
    	public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        
    		System.out.println("返回消息:" + new String(message.getBody()) + ",replyCode:" + replyCode
    				+ ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);
    	}
    
    }

#### producer服务 ####

发送者

测试

import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    @RestController
    public class AckTestController {
        
        @Autowired
        private AckSender ackSender;
    
        @GetMapping("messageack")
        public String messageack() throws InterruptedException {
        
            for (int i = 0; i < 10; i++) {
        
                ackSender.send(i);
            }
             return "messageack";
        }
    
    }

访问 http://127.0.0.1:8080/messageack,consumer控制台如下：

消费者A : User(name=张0, age=19, sex=男)Mon Feb 13 15:59:00 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    消费者A : User(name=张1, age=19, sex=男)Mon Feb 13 15:59:00 CST 2023
    消息消费A成功！
    确认发送ok ,Mon Feb 13 15:59:00 CST 2023,{
        "age":19,"name":"张0","sex":"男"}
    消费者A : User(name=张3, age=19, sex=男)Mon Feb 13 15:59:01 CST 2023
    消息消费A成功！
    消费者A : User(name=张5, age=19, sex=男)Mon Feb 13 15:59:02 CST 2023
    消息消费A成功！
    消费者A : User(name=张6, age=19, sex=男)Mon Feb 13 15:59:03 CST 2023
    消息消费A成功！
    消费者A : User(name=张8, age=19, sex=男)Mon Feb 13 15:59:04 CST 2023
    消息消费A成功！
    消费者A : User(name=张2, age=19, sex=男)Mon Feb 13 15:59:05 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    确认发送ok ,Mon Feb 13 15:59:05 CST 2023,{
        "age":19,"name":"张2","sex":"男"}
    消费者A : User(name=张2, age=19, sex=男)Mon Feb 13 15:59:05 CST 2023
    消息消费A成功！
    消费者A : User(name=张4, age=19, sex=男)Mon Feb 13 15:59:10 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    确认发送ok ,Mon Feb 13 15:59:10 CST 2023,{
        "age":19,"name":"张4","sex":"男"}
    消费者A : User(name=张0, age=19, sex=男)Mon Feb 13 15:59:15 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    确认发送ok ,Mon Feb 13 15:59:15 CST 2023,{
        "age":19,"name":"张0","sex":"男"}
    消费者A : User(name=张0, age=19, sex=男)Mon Feb 13 15:59:15 CST 2023
    消息消费A成功！
    消费者A : User(name=张7, age=19, sex=男)Mon Feb 13 15:59:20 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    确认发送ok ,Mon Feb 13 15:59:20 CST 2023,{
        "age":19,"name":"张7","sex":"男"}
    消费者A : User(name=张9, age=19, sex=男)Mon Feb 13 15:59:25 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    确认发送ok ,Mon Feb 13 15:59:25 CST 2023,{
        "age":19,"name":"张9","sex":"男"}
    消费者A : User(name=张9, age=19, sex=男)Mon Feb 13 15:59:25 CST 2023
    消息消费A成功！
    消费者A : User(name=张4, age=19, sex=男)Mon Feb 13 15:59:30 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    确认发送ok ,Mon Feb 13 15:59:30 CST 2023,{
        "age":19,"name":"张4","sex":"男"}
    消费者A : User(name=张7, age=19, sex=男)Mon Feb 13 15:59:35 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    确认发送ok ,Mon Feb 13 15:59:35 CST 2023,{
        "age":19,"name":"张7","sex":"男"}
    消费者A : User(name=张7, age=19, sex=男)Mon Feb 13 15:59:35 CST 2023
    消息消费A成功！
    消费者A : User(name=张4, age=19, sex=男)Mon Feb 13 15:59:40 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    确认发送ok ,Mon Feb 13 15:59:40 CST 2023,{
        "age":19,"name":"张4","sex":"男"}
    消费者A : User(name=张4, age=19, sex=男)Mon Feb 13 15:59:45 CST 2023
    消息消费B失败:/ by zerojava.lang.ArithmeticException: / by zero
    确认发送ok ,Mon Feb 13 15:59:45 CST 2023,{
        "age":19,"name":"张4","sex":"男"}
    消费者A : User(name=张4, age=19, sex=男)Mon Feb 13 15:59:45 CST 2023
    消息消费A成功！

消息消费A成功！一共出现10次，消息消费B成功，一共出现0次。