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

Question

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

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

工厂setDefaultRequeueRejected（假）；

但我的要求是动态拒绝某些消息，并将某些消息重新排队回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，并对任何想要丢弃的对象抛出amqprejected和dontrequeuerejected。

您可以将所需的逻辑封装在错误处理程序中。

默认错误处理程序正是针对特定的异常列表执行此操作的，如本文所述——您可以插入自定义的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;
    }));

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

如果您没有使用Java 8，请使用new FatalExceptionStrategy（）{…} 。