2

我正在开发一个 Spring 项目，并试图为 RabbitMQ 队列实现带有死字的指数退避。
在此过程中，我创建了一个死信队列和一个死信交换（扇出），并将原始队列的 x-dead-letter-exchange 参数设置为死信交换的名称，并创建了一个带有 ExponentialBackOffPolicy 的 RetryTemplate .
出于测试目的，我的消费者只是通过抛出异常来拒绝它收到的所有消息。

这是我的 RabbitMQConfiguration 类的样子：

@Configuration
@EnableAutoConfiguration
@PropertySource("file:${HOME}/common/config/wave-planning.properties")
public class RabbitMQConfiguration {

    private final static String QUEUE_NAME = "orderPlanQueue";

    private static final String EXCHANGE_NAME = "orderPlanExchange";

    private static final String DL_EXCHANGE_NAME = "deadLetterExchange";

    private static final String DL_QUEUE_NAME = "deadLetterQueue";

    @Value("${rabbitmq.host:localhost}")
    private String host;

    @Value("${rabbitmq.port:5672}")
    private int port;

    @Value("${rabbitmq.user:guest}")
    private String userName;

    @Value("${rabbitmq.password:guest}")
    private String password;

    @Value("${rabbitmq.initial_backoff_interval:1000}")
    private int INITIAL_INTERVAL_IN_MILLISECONDS;

    @Value("${rabbitmq.max_backoff_interval:10000}")
    private int MAX_INTERVAL_IN_MILLISECONDS;

    @Autowired
    OrderPlanService orderPlanService;

    @Bean
    Queue queue() {
        Map<String, Object> qargs = new HashMap<String, Object>();
        qargs.put("x-dead-letter-exchange", DL_EXCHANGE_NAME);
        return new Queue(QUEUE_NAME, false, false, false, qargs);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    FanoutExchange deadLetterExchange() { return new FanoutExchange(DL_EXCHANGE_NAME); }

    @Bean
    Queue deadLetterQueue() { return new Queue(DL_QUEUE_NAME); }

    @Bean
    Binding deadLetterBinding(Queue deadLetterQueue, FanoutExchange deadLetterExchange) {
        return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange);
    }

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

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(userName);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }

    @Bean
    public MessageConverter Jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public AmqpTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());

        RetryTemplate retry = new RetryTemplate();
        ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();

        policy.setInitialInterval(INITIAL_INTERVAL_IN_MILLISECONDS);
        policy.setMultiplier(2);
        policy.setMaxInterval(MAX_INTERVAL_IN_MILLISECONDS);

        retry.setBackOffPolicy(policy);
        template.setRetryTemplate(retry);

        template.setRoutingKey(QUEUE_NAME);
        template.setMessageConverter(Jackson2JsonMessageConverter());
        return template;
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setMessageConverter(Jackson2JsonMessageConverter());
        container.setQueueNames(QUEUE_NAME);
        container.setMessageListener(listenerAdapter);
        container.setDefaultRequeueRejected(false);
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter() {
        return new MessageListenerAdapter(orderPlanService, "consume");
    }
}

消费者的相关部分基本上是这样的：

@Service
@Transactional
public class BaseOrderPlanService implements OrderPlanService {

    ....

    @Override
    public void consume(Object object) {
        throw new IllegalArgumentException("Test");
    }
}

对于自动装配的整数值，使用默认值。
在运行此程序时，我看到交换和队列按预期在 rabbitmq 上创建，并在相关时使用预期的绑定和参数。
但是，当我使用路由键“orderPlanQueue”将消息传递给 orderPlanExchange 时，会导致无限循环，因为消息在队列中被拒绝并重复替换。
另一方面，如果将 IllegalArgumentException 替换为 AmqpRejectAndDontRequeueException，则在第一次拒绝尝试时将消息简单地扔到死信队列中。

如果有人能指出我在这里可能做错了什么，没有应用重试策略，我将不胜感激。

编辑：根据 Artem 的建议使用 StatefulRetryOperationsInterceptor 编写代码。

@Configuration
@EnableAutoConfiguration
@PropertySource("file:${HOME}/common/config/wave-planning.properties")
public class RabbitMQConfiguration {

    private final static String QUEUE_NAME = "orderPlanQueue";

    private static final String EXCHANGE_NAME = "orderPlanExchange";

    private static final String DL_EXCHANGE_NAME = "deadLetterExchange";

    private static final String DL_QUEUE_NAME = "deadLetterQueue";

    @Value("${rabbitmq.host:localhost}")
    private String host;

    @Value("${rabbitmq.port:5672}")
    private int port;

    @Value("${rabbitmq.user:guest}")
    private String userName;

    @Value("${rabbitmq.password:guest}")
    private String password;

    @Value("${rabbitmq.initial_backoff_interval:1000}")
    private int INITIAL_INTERVAL_IN_MILLISECONDS;

    @Value("${rabbitmq.max_backoff_interval:10000}")
    private int MAX_INTERVAL_IN_MILLISECONDS;

    @Autowired
    OrderPlanService orderPlanService;

    @Bean
    Queue queue() {
        Map<String, Object> qargs = new HashMap<String, Object>();
        qargs.put("x-dead-letter-exchange", DL_EXCHANGE_NAME);
        return new Queue(QUEUE_NAME, false, false, false, qargs);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

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

    @Bean
    FanoutExchange deadLetterExchange() { return new FanoutExchange(DL_EXCHANGE_NAME); }

    @Bean
    Queue deadLetterQueue() { return new Queue(DL_QUEUE_NAME); }

    @Bean
    Binding deadLetterBinding(Queue deadLetterQueue, FanoutExchange deadLetterExchange) {
        return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange);
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(userName);
        connectionFactory.setPassword(password);
        return connectionFactory;
    }

    @Bean
    public MessageConverter Jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public AmqpTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());

        /*
        RetryTemplate retry = new RetryTemplate();
        ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();

        policy.setInitialInterval(INITIAL_INTERVAL_IN_MILLISECONDS);
        policy.setMultiplier(2);
        policy.setMaxInterval(MAX_INTERVAL_IN_MILLISECONDS);

        retry.setBackOffPolicy(policy);
        template.setRetryTemplate(retry);
        */

        template.setRoutingKey(QUEUE_NAME);
        template.setMessageConverter(Jackson2JsonMessageConverter());
        return template;
    }

    @Bean
    StatefulRetryOperationsInterceptor interceptor() {
        return RetryInterceptorBuilder.stateful()
                .maxAttempts(4)
                .backOffOptions(INITIAL_INTERVAL_IN_MILLISECONDS, 2, MAX_INTERVAL_IN_MILLISECONDS)
                .build();
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setMessageConverter(Jackson2JsonMessageConverter());
        container.setQueueNames(QUEUE_NAME);
        container.setMessageListener(listenerAdapter);
        container.setAdviceChain(new Advice[] {interceptor()});
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter() {
        return new MessageListenerAdapter(orderPlanService, "consume");
    }

}
4

1 回答 1

2

重试策略RabbitTemplate与 DLQ/DLX 完全无关。那是针对消费者方面的。

请参阅此处参考手册中的差异：

您现在可以配置RabbitTemplate以使用 aRetryTemplate来帮助处理代理连接问题。

在这里：

要限制客户端重新交付的次数，一种选择是StatefulRetryOperationsInterceptor在侦听器的建议链中。

因此，您必须重新考虑您的逻辑并将重试功能置于SimpleMessageListenerContainer定义中。

于 2017-07-10T19:41:29.740 回答