rabbitMq消息处理两种方式（手动和自动）

桃扇骨 2022-08-31 11:02 181阅读 0赞

> 前提：`TestDirectQueueProviderOne 绑定死信交换机`

**省略怎么创建死信队列和绑定私信队列**

# `消息变成死信有以下几种情况` #

消息变成死信之后，如果绑定了死信队列，消息会进入死信队列

>  *  消息被拒绝(basic.reject / basic.nack)，并且requeue = false
>  *  消息TTL过期
>  *  队列达到最大长度

# 手动ack #

1.  application.yml

listener:
          simple:
            acknowledge-mode: manual   # 手动确认消息

1.  消费端
    
    消息处理：
    
        @RabbitListener(queues = "TestDirectQueueProviderOne")//监听的队列名称 TestDirectQueueProvider
            public void process(Map testMessage, Message message, Channel channel) throws IOException { 
                long deliveryTag = message.getMessageProperties().getDeliveryTag();
                    System.out.println("DirectReceiver1消费者收到消息 : " + testMessage.toString());
                    System.out.println(LocalDateTime.now());
                    int a = 1 / 0;
                    //确认签收
                    channel.basicAck(deliveryTag, true);
        
        //☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆【以下适用于手动ack,application.yml: acknowledge-mode: manual】☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆
        // ###################以下是捕获异常之后重试一次，不行就丢弃（如果绑定死信队列就会到死信队列）##############################
                    System.out.println("=====================异常了========================");
                    if (message.getMessageProperties().getRedelivered()) { 
                        System.out.println("================消息已重复处理失败,拒绝再次接收======================");
                        // 拒绝消息，requeue=false 表示不再重新入队，如果配置了死信队列则进入死信队列
                        channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                    } else { 
                        System.out.println("====================消息即将再次返回队列处理=========================");
                        // requeue为是否重新回到队列，true重新入队
                        channel.basicReject(message.getMessageProperties().getDeliveryTag(),  true);
        
                    }
            }

# 自动ack #

1.  application.yml

listener:
          simple:
            acknowledge-mode: auto   # 手动确认消息
            retry:
              #enabled：开启失败重试
              enabled: true
              #第一次重试的间隔时长
              initial-interval: 5000ms
              #最长重试间隔，超过这个间隔将不再重试
              max-interval: 300000ms
              #下次重试间隔的倍数，此处是2即下次重试间隔是上次的2倍
              multiplier: 2
              max-attempts: 3 
            default-requeue-rejected: true

1.  消费端
    
    消息处理：

@RabbitListener(queues = "TestDirectQueueProviderOne")//监听的队列名称 TestDirectQueueProvider
        public void process(Map testMessage, Message message, Channel channel) throws IOException { 
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
    // try { 
                System.out.println("DirectReceiver1消费者收到消息 : " + testMessage.toString());
                System.out.println(LocalDateTime.now());
                int a = 1 / 0;
                //确认签收
                channel.basicAck(deliveryTag, true);
    // } catch (Exception e) { 
    
    //☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆【以下适用于手动ack,application.yml: acknowledge-mode: auto,default-requeue-rejected: false】☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆☆
    // throw e; //如果使用默认的重试机制的话，不需要try-catch，即使try-catch也要再抛出异常，不抛出重试会失效
    
        }

# 监听死信队列 #

@RabbitListener(queues = "dlx.queue")
        public void process(Map testMessage, Message message, Channel channel) throws IOException { 
            System.out.println("#####################进入死信队列######################");
            System.out.println(testMessage);
            //对消息进行业务处理....，比如30min未支付修改订单状态为超时
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }