RabbitMQ学习（五）-- 消息Consumer ack

以你之姓@ 2022-12-29 04:58 141阅读 0赞

### 目录 ###

*  一、修改配置文件
 *  二、实现ChannelAwareMessageListener接口
 *  三、测试结果
 *  四、总结

# 一、修改配置文件 #

spring:
      rabbitmq:
        listener:
          type: direct
          direct:
            acknowledge-mode: manual #手动模式

# 二、实现ChannelAwareMessageListener接口 #

package com.wayne.listener;
    
    import com.rabbitmq.client.Channel;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    
    import static com.wayne.core.config.RabbitConfig.QUEUE_NAME;
    
    @Slf4j
    @Component
    public class RabbitMQConsumerQueue implements ChannelAwareMessageListener { 
    
    
    
    
    
    // @RabbitListener(queues = QUEUE_NAME)
    // public void rabbitMQConsumerListener(Message message){ 
    // log.info("当前队列信息为"+message);
    // }
    
    
        @Override
        @RabbitListener(queues = QUEUE_NAME)
        public void onMessage(Message message, Channel channel) throws Exception { 
    
            Thread.sleep(1000);
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            try{ 
    
                //1.接收转换参数
                log.info(new String(message.getBody()));
                //2.接收业务处理
                log.info("业务处理");
                int i = 3/0;
                //3.手动签收
                /** * deliveryTag：发送状态 * multiple:允许多条消息同时被签收 */
                channel.basicAck(deliveryTag,true);
            }catch (Exception e){ 
                //4.拒绝签收
                /** * deliveryTag：发送状态 * multiple:允许多条消息同时被签收 * requeue：是否重新放入队列 */
                channel.basicNack(deliveryTag,true,true);
            }
        }
    }

# 三、测试结果 #

2020-12-20 16:13:14,785 INFO (AbstractConnectionFactory.java:511)- Created new connection: rabbitConnectionFactory#6b6b3572:0/SimpleConnection@3f68a7f8 [delegate=amqp://guest@10.211.55.7:5672/, localPort= 65251]
    2020-12-20 16:13:14,817 INFO (DirectMessageListenerContainer.java:422)- Container initialized for queues: [boot_queue]
    2020-12-20 16:13:14,823 INFO (DirectMessageListenerContainer.java:688)- SimpleConsumer [queue=boot_queue, consumerTag=amq.ctag-N012hlEGPbzORk3Z6PtMsQ identity=49c44937] started
    2020-12-20 16:13:14,825 INFO (StartupInfoLogger.java:61)- Started SpringBootDemo in 27.253 seconds (JVM running for 32.789)
    2020-12-20 16:13:15,835 INFO (RabbitMQConsumerQueue.java:39)- hello spring boot rabbitMQ!~~~~~~~
    2020-12-20 16:13:15,837 INFO (RabbitMQConsumerQueue.java:41)- 业务处理
    2020-12-20 16:13:16,844 INFO (RabbitMQConsumerQueue.java:39)- hello spring boot rabbitMQ!~~~~~~~
    2020-12-20 16:13:16,845 INFO (RabbitMQConsumerQueue.java:41)- 业务处理
    2020-12-20 16:13:17,853 INFO (RabbitMQConsumerQueue.java:39)- hello spring boot rabbitMQ!~~~~~~~
    2020-12-20 16:13:17,854 INFO (RabbitMQConsumerQueue.java:41)- 业务处理
    2020-12-20 16:13:18,860 INFO (RabbitMQConsumerQueue.java:39)- hello spring boot rabbitMQ!~~~~~~~

# 四、总结 #

实现步骤

1.  设置手动模式 spring.rabbitmq.listener.direct.acknowledge-mode = manual
2.  消费者监听器实现ChannelAwareMessageListener接口
3.  使用channel.basicAck(deliveryTag,true);手动签收消息
4.  使用channel.basicNack(deliveryTag,true,true);拒绝签收消息，并放回原消息队列。