java - Spring AMQP - 重复消息

Question

我正在处理一个大流量流 ~ 每秒 500 多条消息，使用 SimpleMessageListenerContainer 和 10 个并发消费者从 Spring AMQP+Rabbit 消耗数据，我必须每 15 分钟对 Db 进行一些检查并重新加载某些属性对于处理，这是通过每 15 分钟触发一次的 quartz 触发器完成的，停止 SimplelistenerContainer，执行必要的工作并再次启动容器。

当应用程序启动时一切正常，当触发器触发并且容器重新启动时，我看到同一条消息被多次传递，这导致了很多重复。消费者没有抛出异常。

消息监听器

 class RoundRobinQueueListener implements MessageListener {

@Override
public void onMessage(Message message) { //do processing
     }

  }

在应用启动期间设置并行消费者并启动消费者

final SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
       RoundRobinQueueListener roundRobinListener = RoundRobinQueueListener.class.newInstance();
        messageListenerContainer.setQueueNames(queueName);
        messageListenerContainer.setMessageListener(roundRobinListener);
        messageListenerContainer.setConcurrentConsumers(10);
        messageListenerContainer.setChannelTransacted(true);

quartz 触发器

    void execute(JobExecutionContext context) throws JobExecutionException {
    messageListenerContainer.stop()
    //Do db task, other processing
    messageListenerContainer.start()
    }

Answer 1

看起来您的消息现在已被消费者确认。如果你没有使用自动确认模式，你需要自己确认消息(这也可以在 SimpleMessageListenerContainer 中配置)。否则，代理会假定消息未成功处理并尝试再次传递它。