如果未处理，则弹簧集成 amqp 拒绝消息

Question

我需要解决这种情况。我有两个 amqp 消费者设置来获取一条消息。

@Bean
public IntegrationFlow jmsPrimaryFlow() {
    return IntegrationFlows.from(
        Amqp.inboundGateway(
            taskManager().getPrimaryMessageListenerContainer()).errorChannel(errorChannel())
        )
        .channel(taskChannel())
        .get();
}

@Bean
public IntegrationFlow jmsSecondaryFlow() {
    return IntegrationFlows.from(
        Amqp.inboundGateway(
            taskManager().getSecondaryMessageListenerContainer()).errorChannel(errorChannel())
            .autoStartup(false)
        )
        .channel(taskChannel())
        .get();
}

taskChannel 是 queuechannel，但一次只允许一条消息消费，因此没有并行处理。 如果另一条消息花费太长时间才能继续，我如何在超时后拒绝一条消息。 所以这条消息将返回到队列以由另一个节点继续？只是我的意思是这两个消费者预取两条消息，但一次只能处理一条消息，所以如果第一个消息需要很长时间才能继续，如何释放第二条预取消息。

Answer 1

你的问题不清楚。您可以在队列通道上设置容量限制（例如 1）并在网关上设置 sendTimeout。然后，如果队列已满，则在超时后尝试添加消息将失败。但是，在这种情况下使用队列通道是危险的——如果服务器出现故障，您可能会丢失消息，因为消息在存入队列后立即被确认。

如果你改用RendezvousChannel，生产者将阻塞等待消费者接收消息。

但请记住，如果服务器在切换后崩溃，即使是这一条消息也可能丢失。