rabbitmq_mq手动ack的配置_rabbitmq手动ack

由于公司配置的mq是自动ack的，所以导致有时候数据丢失没有保存数据，所以想改为手动ack的方式。
首先按照我的理解，手动ack和自动ack的含义分别是：
手动ack：
消费者收到消息之后，等到业务流程走完，确当数据已经落地，自己手动告诉mq服务端我收到消息了，然后服务端再把消息销毁。
自动ack：
不管你数据有没有落地，消费者一旦收到消息就自动ack告诉服务端我收到消息了，然后服务端清理数据。

公司的项目是ssh的，在xml中配置mq我还是比较生疏的，参考网上的资料，把mq配成手动ack的。

<!-- 配置手动ack -->
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="manual">
    <!--    <rabbit:listener queues="hello-java-queue" ref="ChannelConsumer"/>-->
</rabbit:listener-container>
1
2
3
4

其中注释掉的代码是网上参考的资料说要绑定一个监听器，但是我直接采用@RabbitListener的方法去监听，所以不需要这个监听器。
当我在代码中执行ack的时候

@RabbitListener(queues = "test-java-queue")
private void commonMethod(List<HashMap<String, String>> maps, Message message, Channel channel) {
    try {
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
//            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    } catch (IOException e) {
        e.printStackTrace();
    }
    System.out.println("333333333333333333");
}
1
2
3
4
5
6
7
8
9
10

控制台爆出了错误

ERROR o.s.a.r.c.CachingConnectionFactory - Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - unknown delivery tag 1, class-id=60, method-id=80)
1

其中报错的原因在这里。

unknown delivery tag 1
1

由于每条channel都会有属于自己的tag，消费者端跟服务器连接的通道里面靠唯一的tag来辨识每条channel，这里报错的意思是，对于这条通道的channel的tag是不存在的。
按照常理来说，我们配置了手动ack，应该会等我们ack之后这条通道才会关闭，同时销毁tag。但是这个时候tag被销毁了，那么就是意味着有人替我们ack了。
参考网上的资料，在配置
SimpleRabbitListenerContainerFactory
的时候，如果配置了
setMessageConverter
项，那么mq会自己帮我们ack，网上示例的配置代码如下

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setMessageConverter(new Jackson2JsonMessageConverter());
    return factory;
}
1
2
3
4
5
6
7

这里如果配置了

factory.setMessageConverter(new Jackson2JsonMessageConverter());
1

这一项，那么mq就会变成自动ack。所以在这里还需要多配置一个factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
即配置完的配置文件为

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setMessageConverter(new Jackson2JsonMessageConverter());
    factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    return factory;
    }
1
2
3
4
5
6
7
8

但是公司项目是ssh的，在xml里配置，配置文件如下

<bean id="rabbitListenerContainerFactory" class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
    <!-- 配置手动ack -->
    <property name="acknowledgeMode" value="MANUAL"/>
</bean>
1
2
3
4

只配置了其中需要配置的一条，其它配置文件因保密协议不能在公共场合发布。
至此，mq的手动ack配置完成。