如何在Spring AMQP中使用Ack或Nack

Question

我是SpringAMQP的新手。我有一个应用程序是生产者发送消息给另一个应用程序是消费者。

一旦消费者收到消息，我们将对数据进行验证。

如果数据是正确的，我们必须确认，消息应该从队列中删除。如果数据不正确，我们必须对数据进行NACK（否定确认），以便它在RabbitMQ中重新排队。

我偶然发现

**工厂。setDefaultRequeueRejected（假）** （它根本不会重新发出消息）

**工厂。setDefaultRequeueRejected（真）** （发生异常时将重新发出消息）

但在我的情况下，我会根据确认确认信息。然后它应该删除该消息。如果是NACK，则重新获取消息。

我在RabbitMQ网站上看过

AMQP规范定义了基本的。拒绝方法，该方法允许客户端拒绝单个已传递的消息，指示代理放弃或重新获取这些消息

如何实现上述场景？请给我举几个例子。

我尝试了一个小程序

       logger.info("Job Queue Handler::::::::::" + new Date());
        try {

        }catch(Exception e){

            logger.info("Activity Object Not Found Exception so message should be Re-queued the Message::::::::::::::");

        }

        factory.setErrorHandler(new ConditionalRejectingErrorHandler(cause ->{
            return cause instanceof XMLException;
        }));

消息没有为不同的异常工厂重新排队。setDefaultRequeueRejected（真）

09:46:38854错误[stderr]（SimpleAsyncTaskExecutor-1）组织。activiti。发动机ActivitiObjectNotFoundException:没有使用键“WF89012”部署进程

09:46:39102从错误队列接收到的信息[com.example.bip.rabbitmq.handler.ErrorQueueHandler]（SimpleAsynctaskeExecutor-1）：{Error=无法提交JPA事务；嵌套异常为javax.persistence.RollbackException:标记为rollbackOnly的事务}




Answer 1

请参阅文档。

默认情况下，如果侦听器正常退出，容器将装入消息（导致消息被删除），如果侦听器抛出异常，容器将拒绝（并请求）它。

如果侦听器（或错误处理程序）抛出一个AmqpRejectAndDontRepeueExctive，则会重写默认行为并丢弃消息（或路由到DLX/DLQ（如果这样配置）-容器调用basicReject（false）而不是basicReject（true）< /code>。

因此，如果验证失败，抛出一个AmqpRejectAndDontRequeueException。或者，使用自定义错误处理程序配置侦听器，将异常转换为AmqpRejectAndDontRequeueException。

这在本答复中有所描述。

如果你真的想自己负责确认，请将确认模式设置为MANUAL，并使用ChannelAwareMessageListener，如果你使用的是@RabbitListener，则使用此技术。

但大多数人只是让容器来处理事情（一旦他们明白了发生了什么）。通常，使用手动确认是为了特殊的使用情况，例如延迟确认或提前确认。

编辑

我给你指出的答案中有一个错误（现已修复）；你必须看看ListenerExefftionFailedExctive的原因。我刚刚测试了这个，它像预期的那样工作...

@SpringBootApplication
public class So39530787Application {

    private static final String QUEUE = "So39530787";

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = SpringApplication.run(So39530787Application.class, args);
        RabbitTemplate template = context.getBean(RabbitTemplate.class);
        template.convertAndSend(QUEUE, "foo");
        template.convertAndSend(QUEUE, "bar");
        template.convertAndSend(QUEUE, "baz");
        So39530787Application bean = context.getBean(So39530787Application.class);
        bean.latch.await(10, TimeUnit.SECONDS);
        System.out.println("Expect 1 foo:"  + bean.fooCount);
        System.out.println("Expect 3 bar:"  + bean.barCount);
        System.out.println("Expect 1 baz:"  + bean.bazCount);
        context.close();
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setErrorHandler(new ConditionalRejectingErrorHandler(
                t -> t instanceof ListenerExecutionFailedException && t.getCause() instanceof FooException));
        return factory;
    }

    @Bean
    public Queue queue() {
        return new Queue(QUEUE, false, false, true);
    }
    private int fooCount;

    private int barCount;

    private int bazCount;

    private final CountDownLatch latch = new CountDownLatch(5);

    @RabbitListener(queues = QUEUE)
    public void handle(String in) throws Exception {
        System.out.println(in);
        latch.countDown();
        if ("foo".equals(in) && ++this.fooCount < 3) {
            throw new FooException();
        }
        else if ("bar".equals(in) && ++this.barCount < 3) {
            throw new BarException();
        }
        else if ("baz".equals(in)) {
            this.bazCount++;
        }
    }

    @SuppressWarnings("serial")
    public static class FooException extends Exception { }

    @SuppressWarnings("serial")
    public static class BarException extends Exception { }

}

结果：

Expect 1 foo:1
Expect 3 bar:3
Expect 1 baz:1