如何使用 SimpleMessageListenerContainer bean 从 RabbitMq 队列中破坏性地读取消息

Question

我需要帮助来实现 SimpleMessageListenerContainer，它会在消息接收者读取后立即以非事务方式删除消息。

在我的情况下，无论事务成功或不成功，放入队列的消息都会挂在某处（不在队列中），并在从队列读取的每个操作中重复处理。因此，放入队列的所有其他消息仍然无法访问，并且每次仅重新处理第一个消息。

另一个奇怪的事情是，我看不到队列中的消息登陆/排队，即在 Rabbit 管理控制台上队列深度永远不会改变，而只是消息速率在每次写入队列时都会发生跳跃。

下面是我的 Java 配置的代码 sn-p。如果有人可以在这里指出错误，那将有很大帮助：-

@Configuration
@EnableJpaRepositories(basePackages={"com.xxx.muppets"})
public class MuppetMessageConsumerConfig {

private ApplicationContext applicationContext;
@Value("${rabbit.queue.name}")
private String queueName;

@Autowired
public void setApplicationContext(ApplicationContext applicationContext) {
    this.applicationContext = applicationContext;
}

@Bean
Queue queue() {
    return new Queue(queueName, false);
}

@Bean
TopicExchange exchange() {
    return new TopicExchange("spring-boot-exchange");
}

@Bean
Binding binding(Queue queue, TopicExchange exchange) {
    return BindingBuilder.bind(queue).to(exchange).with(queueName);
}

@Bean
MuppetMsgReceiver muppetMsgReceiver(){
    return new MuppetMsgReceiver();
}

@Bean
MessageListenerAdapter listenerAdapter(MuppetMsgReceiver muppetMsgReceiver){
    return new MessageListenerAdapter(muppetMsgReceiver, "receiveMessage");
}

@Bean
SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setAcknowledgeMode(NONE);
    container.setConnectionFactory(connectionFactory);
    container.setQueueNames(queueName);
    container.setMessageListener(listenerAdapter);
    return container;
}
}

我的消息接收类如下：

public class MuppetMsgReceiver {

private String muppetMessage;

private CountDownLatch countDownLatch;

public MuppetMsgReceiver() {
    this.countDownLatch = new CountDownLatch(1);
}

public MuppetMsgReceiver(CountDownLatch latch) {
    this.countDownLatch = latch;        
    CountDownLatch getCountDownLatch() {
    return countDownLatch;
}

public void receiveMessage(String receivedMessage) {
    countDownLatch.countDown();
    this.muppetMessage = receivedMessage;
}

public String getMuppetMessage() {
    return muppetMessage;
}
}

此完整代码基于 Spring 的 getting started example，但由于队列中的非破坏性读取而没有帮助。

Answer 1

AcknowledgeMode=NONE 表示代理在消息发送 给消费者后立即确认消息，而不是在消费者收到消息时确认，因此您观察到的行为对我来说毫无意义。使用这种确认模式是很不寻常的，但是消息一发送就会消失。这可能是您的代码有问题。 TRACE 日志记录将帮助您了解正在发生的事情。