对 RabbitMQ 队列的否定确认以使用 Spring AMQP 重新排队消息

Question

我有一个应用程序，它使用 spring AMQP 为其他应用程序消费和生成消息。我有一个场景，其中发生了一些异常，我需要重新排队回到 RabbitMQ。对于一些我需要忽略的异常（基本上我需要忽略不需要重新排队的消息）

目前在下面的代码中，我已将配置设置为

factory.setDefaultRequeueRejected(false);

但我的要求是动态拒绝某些消息，并重新排队返回 RabbitMQ 以获取某些消息。

请建议

@Bean(name="rabbitListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter();
        DefaultClassMapper classMapper = new DefaultClassMapper();
        Map<String, Class<?>> idClassMapping = new HashMap<String, Class<?>>();
        idClassMapping.put(Constants.JOB_TYPE_ID_, JobListenerDTO.class);
        classMapper.setIdClassMapping(idClassMapping);
        messageConverter.setClassMapper(classMapper);
        factory.setMessageConverter(messageConverter);
        factory.setDefaultRequeueRejected(false);
        factory.setReceiveTimeout(10L);
        return factory;
    }

Answer 1

你不能那样做（默认为 false）。

要有选择地执行此操作，您必须设置defaultRequeueRejected为true并抛出AmqpRejectAndDontRequeueRejected您想要丢弃的任何内容。

您可以将所需的逻辑封装在ErrorHandler.

默认错误处理程序完全针对特定的异常列表执行此操作，如此处所述- 您可以注入自定义FatalExceptionStrategy.

但是对于有条件拒绝，defaultRequeueRejected必须是true。

编辑

factory.setErrorHandler(new ConditionalRejectingErrorHandler(t -> {
        Throwable cause = t.getCause();
        return cause instanceof MessageConversionException
                || cause instanceof org.springframework.messaging.converter.MessageConversionException
                || cause instanceof MethodArgumentNotValidException
                || cause instanceof MethodArgumentTypeMismatchException
                || cause instanceof NoSuchMethodException
                || cause instanceof ClassCastException
                || cause instanceof MyBadXMLException;
    }));

这将添加MyBadXMLException到标准列表中。

如果您使用的不是 Java 8，请使用new FatalExceptionStrategy() {...}.