Spring AMQP - 使用死信机制和TTL重新排队消息 [英] Spring AMQP - Message re queuing using dead letter mechanism with TTL

问题描述

它像休斯顿我们在这里遇到问题，我需要在首次尝试处理事件失败后安排/延迟消息5分钟。
我已经在这种情况下实现了死信交换。

Its like "Houston we have a problem here" where I need to schedule/delay a message for 5 minutes after it fails on the first attempt to process an event. I have implemented dead letter exchange in this scenario.

失败的消息，路由到DLX - >重试队列并在之后返回工作队列另一次尝试的TTL为5分钟。

The messages on failing, route to the DLX --> Retry Queue and comes back to work queue after a TTL of 5 minutes for another attempt.

以下是我使用的配置：

public class RabbitMQConfig {
    @Bean(name = "work")
    @Primary
    Queue workQueue() {
        return new Queue(WORK_QUEUE, true, false, false, null);
    }

    @Bean(name = "workExchange")
    @Primary
    TopicExchange workExchange() {
        return new TopicExchange(WORK_EXCHANGE, true, false);
    }

    @Bean
    Binding workBinding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(workQueue()).to(workExchange()).with("#");
    }

    @Bean(name = "retryExchange")
    FanoutExchange retryExchange() {
        return new FanoutExchange(RETRY_EXCHANGE, true, false);
    }

    @Bean(name = "retry")
    Queue retryQueue() {
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-dead-letter-exchange", WORK_EXCHANGE);
        args.put("x-message-ttl", RETRY_DELAY); //delay of 5 min
        return new Queue(RETRY_QUEUE, true, false, false, args);
    }

    @Bean
    Binding retryBinding(Queue queue,FanoutExchange exchange) {
        return BindingBuilder.bind(retryQueue()).to(retryExchange());
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }

    @Bean
    Consumer receiver() {
        return new Consumer();
    }

    @Bean
    MessageListenerAdapter listenerAdapter(Consumer receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }
}

Producer.java：

Producer.java:

@GetMapping(path = "/hello")
public String sayHello() {
    // Producer operation

        String messages[];
        messages = new String[] {" hello "};

    for (int i = 0; i < 5; i++) {
        String message = util.getMessage(messages)+i;

        rabbitTemplate.convertAndSend("WorkExchange","", message);
       System.out.println(" Sent '" + message + "'");
    }
    return "hello";
}

Consumer.java：

Consumer.java:

public class Consumer {

    @RabbitListener(queues = "WorkQueue")
    public void receiveMessage(String message, Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) Long tag) throws IOException, InterruptedException {

        try {

            System.out.println("message to be processed: " + message);
            doWorkTwo(message);
            channel.basicAck(tag, false);

        } catch (Exception e) {
            System.out.println("In the exception catch block");
            System.out.println("message in dead letter exchange: " + message);
            channel.basicPublish("RetryExchange", "", null, message.getBytes());

        }

    }

    private void doWorkTwo(String task) throws InterruptedException {

        int c = 0;
        int b = 5;
        int d = b / c;

    }

}

这是在我的情景中使用死信交换的正确方法和在RETRY QUEUE中等待一次5分钟后的，在第二次尝试时它不会在RETRY QUEUE中等待5分钟（我已经提到TTL为5分钟） ）并立即移至工作队列。

Is it the correct way to use a dead letter exchange for my scenario and after waiting once in the RETRY QUEUE for 5 min, on the second time attempt it does not wait for 5 min in the RETRY QUEUE (I have mentioned TTL as 5 min) and moves to the WORK QUEUE immediately.

我通过点击localhost：8080 / hello url运行此应用程序。

I am running this application by hitting localhost:8080/hello url.

这是我的更新配置。

Here is my updated configuration.

RabbitMQConfig.java：

RabbitMQConfig.java:

@EnableRabbit
public class RabbitMQConfig {

    final static String WORK_QUEUE = "WorkQueue";
    final static String RETRY_QUEUE = "RetryQueue";
    final static String WORK_EXCHANGE = "WorkExchange"; // Dead Letter Exchange
    final static String RETRY_EXCHANGE = "RetryExchange";
    final static int RETRY_DELAY = 60000; // in ms (1 min)

    @Bean(name = "work")
    @Primary
    Queue workQueue() {
         Map<String, Object> args = new HashMap<String, Object>();
         args.put("x-dead-letter-exchange", RETRY_EXCHANGE);
        return new Queue(WORK_QUEUE, true, false, false, args);
    }

    @Bean(name = "workExchange")
    @Primary
    DirectExchange workExchange() {
        return new DirectExchange(WORK_EXCHANGE, true, false);
    }

    @Bean
    Binding workBinding(Queue queue, DirectExchange exchange) {
        return BindingBuilder.bind(workQueue()).to(workExchange()).with("");
    }

    @Bean(name = "retryExchange")
    DirectExchange retryExchange() {
        return new DirectExchange(RETRY_EXCHANGE, true, false);
    }

    // Messages will drop off RetryQueue into WorkExchange for re-processing
    // All messages in queue will expire at same rate
    @Bean(name = "retry")
    Queue retryQueue() {
        Map<String, Object> args = new HashMap<String, Object>();
        //args.put("x-dead-letter-exchange", WORK_EXCHANGE);
        //args.put("x-message-ttl", RETRY_DELAY);
        return new Queue(RETRY_QUEUE, true, false, false, null);
    }

    @Bean
    Binding retryBinding(Queue queue, DirectExchange exchange) {
        return BindingBuilder.bind(retryQueue()).to(retryExchange()).with("");
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setDefaultRequeueRejected(false);
        /*factory.setAdviceChain(new Advice[] {
                org.springframework.amqp.rabbit.config.RetryInterceptorBuilder
                        .stateless()
                        .maxAttempts(2).recoverer(new RejectAndDontRequeueRecoverer())
                        .backOffOptions(1000, 2, 5000)
                        .build()
        });*/
        return factory;
    }

    @Bean
    Consumer receiver() {
        return new Consumer();
    }

    @Bean
    MessageListenerAdapter listenerAdapter(Consumer receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }
}

Consumer.java：

Consumer.java:

public class Consumer {

    @RabbitListener(queues = "WorkQueue")
    public void receiveMessage(String message, Channel channel,
            @Header(AmqpHeaders.DELIVERY_TAG) Long tag,
            @Header(required = false, name = "x-death") HashMap<String, String> xDeath)
            throws IOException, InterruptedException {

        doWorkTwo(message);
        channel.basicAck(tag, false);
    }

    private void doWorkTwo(String task) {
        int c = 0;
        int b = 5;
        if (c < b) {
            throw new AmqpRejectAndDontRequeueException(task);
        }
    }
}

推荐答案

如果您拒绝该消息以便代理将其路由到DLQ，您可以检查 x-death 标头。在这种情况下，我有一个TTL为5秒的DLQ，来自主队列的消息的消费者拒绝它;代理将其路由到DLQ，然后它到期并被路由回主队列 - x-death 标题显示重新路由操作的数量：

If you reject the message so the broker routes it to a DLQ, you can examine the x-death header. In this scenario, I have a DLQ with a TTL of 5 seconds and the consumer of the message from the main queue rejects it; the broker routes it to the DLQ, then it expires and is routed back to the main queue - the x-death header shows the number of re-routing operations:

这篇关于Spring AMQP - 使用死信机制和TTL重新排队消息的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！