代码之家 › 专栏 › 技术社区 › sea

SpringAMQP延迟

rabbitmq-exchange spring-rabbit spring-amqp rabbitmq

1

sea · 技术社区 · 9 年前

如果服务不可用或抛出异常,我会调用Webservice,我会将所有请求存储到RabbitMQ队列中,并不断重试服务调用,直到它成功执行。如果服务继续抛出错误或不可用,rabbitMQ侦听器将继续循环。(含义侦听器检索消息,并在出现错误时进行服务调用,重新对消息进行排队)

我使用MessagePostProcessor将循环限制到X小时,但我想在消息级别和每次尝试访问服务时启用延迟。例如,第一次尝试3000毫秒延迟,第二次尝试6000毫秒,以此类推,直到我尝试x次。

你能告诉我一些关于这个的想法吗?

2 回复 | 直到 9 年前

1

Artem Bilan 9 年前

嗯,你这样做是不可能的。

消息重新排队与事务回调完全类似,系统返回到异常前的状态。因此,您肯定不能修改消息以返回队列。

也许你得看看 Spring Retry

在中查看更多信息 Reference Manual .

2

1

sea 9 年前

I added CustomeMessage delay exchange

    @Bean
    CustomExchange delayExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange("delayed-exchange", "x-delayed-message", true, false, args);
    }

Added MessagePostProcessor

  if (message.getMessageProperties().getHeaders().get("x-delay") == null) {
            message.getMessageProperties().setHeader("x-delay", 10000);
        } else {

            Integer integer = (Integer) message.getMessageProperties().getHeaders().get("x-delay");
            if (integer < 60000) {
                integer = integer + 10000;
                message.getMessageProperties().setHeader("x-delay", integer);
            }
        }

First time it delays 30 seconds and adds 10seconds each time till it reaches 600 seconds.This should be configurable.

And finally send the message to 

    rabbitTemplate.convertAndSend("delayed-exchange", queueName,message, rabbitMQMessagePostProcessor);