当前位置:   article > 正文

rabbitmq_mq手动ack的配置_rabbitmq手动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配置完成。

声明:本文内容由网友自发贡献,转载请注明出处:【wpsshop】
推荐阅读
相关标签
  

闽ICP备14008679号