boot - rabbitmq 死信队列

类型

• 消息过期成为死信
• 队列设置 ttl 成为死信

生产者

@RestController
public class IndexController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private static final String exBiz = "ex.exBiz";
    private static final String rkBiz = "rkBiz";

    @GetMapping("/send/deadMsg/{msg}")
    public String send(@PathVariable String msg) {
        rabbitTemplate.convertAndSend(exBiz, rkBiz, MessageBuilder.withBody(msg.getBytes()).build());
        System.out.println("send : " + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(LocalDateTime.now()) +": "+ msg);
        return "success";
    }

    @GetMapping("/send/deadMsg/timeout/{msg}")
    public String sendWithExpire(@PathVariable String msg) {
        rabbitTemplate.convertAndSend(exBiz, rkBiz, MessageBuilder.withBody(msg.getBytes()).setExpiration("20000").build());
        System.out.println("send : " + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(LocalDateTime.now()) +": "+ msg);
        return "success";
    }
}

消费者

@Component
public class Consumer {


    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "${ex.dead:ex.dead}"),
            value = @Queue(value = "${q.dead:q.dead}", durable = "true"),
            key = "dead"
    ))
    public void consumer(String message) {
        System.out.println("dead queue recv: " + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(LocalDateTime.now()) + ": " + message);
    }
}

application.properties 配置

spring.rabbitmq.addresses=192.xxx.12
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

队列设置 ttl 成为死信

@Configuration
public class DeadQueueConfig {

    private static final String exBiz = "ex.exBiz";
    private static final String queueBiz = "q.queueBiz";
    private static final String rkBiz = "rkBiz";

    private static final String rkDead = "dead";

    @Value("${ex.dead:ex.dead}")
    private String exDead;



    /**
     * 业务交换机
     *
     * @return
     */
    @Bean
    public Exchange exBiz() {
        return ExchangeBuilder.directExchange(exBiz).durable(true).build();
    }

    /**
     * 业务队列绑定死信队列
     *
     * @return
     */
    @Bean
    public Queue queueBiz() {
        return QueueBuilder
                .durable(queueBiz)
                .deadLetterExchange(exDead)
                .deadLetterRoutingKey(rkDead)
                // 在队列上设置 过期时间
                .ttl(20000)  //10秒 毫秒
                .build();
    }

    @Bean
    public Binding cancelBinding(@Qualifier("queueBiz") Queue queue,@Qualifier("exBiz") Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(rkBiz).noargs();
    }
}

启动后可以在管理台看见:

可以发现 业务队列被标记成 死信队列

测试

访问: http://localhost:8080/send/deadMsg/hello

结果:

send : 2021-11-10 12:53:28: hello
dead queue recv: 2021-11-10 12:53:48: hello

ttl 消息

注释掉 队列上的 ttl 时间

 @Bean
public Queue queueBiz() {
    return QueueBuilder
            .durable(queueBiz)
            .deadLetterExchange(exDead)
            .deadLetterRoutingKey(rkDead)
            // 在队列上设置 过期时间
            //.ttl(20000)  //10秒 毫秒
            .build();
}

访问: http://localhost:8080/send/deadMsg/timeout/hello

结果:

ttl msg send : 2021-11-10 13:02:46: hello
dead queue recv: 2021-11-10 13:03:06: hello