RabbitMq--延迟队列(死信实现)_rabbitadmin 延迟

RabbitMq–延迟队列(死信实现)

1、 环境

• java 8
• Spring Boot 2.26
• RabbitMq 3.5.4

2、架构图

3、代码

//配置
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: xxx
    password: xxx
    virtual-host: xxx
#    publisher-confirms: true
    publisher-confirm-type: correlated # 发送方确认
    publisher-returns: true  # 交换机找不到队列回调
    listener:
      simple:
        acknowledge-mode: manual # 开启手动ack
        prefetch: 100  # 消费端同时处理的请求个数限制默认250个
//交换机队列等
@Configuration
public class RabbitConfig {
    @Autowired
    private CachingConnectionFactory connectionFactory;
    private final static String DELAY_EXCHANGE = "delay_exchange";
    private final static String DELAY_QUEUE = "delay_queue";
    private final static String DLX_EXCHANGE = "dlx_exchange";
    private final static String DLX_QUEUE = "dlx_queue";
    @Bean
    public RabbitAdmin rabbitadmin(CachingConnectionFactory connectionFactory){
       return new RabbitAdmin(connectionFactory);
    }
    @Bean
    public RabbitTemplate rabbitTemplate(){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //消息转换器
        rabbitTemplate.setMessageConverter(converter());
        //生产这confirm
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if(b){
                    log.info("消息发送成功 ->{}",correlationData.getReturnedMessage());
                }else{
                    log.error("消息->{} 发送失败原因-> {}",correlationData.getReturnedMessage(),s);
                }
            }
        });
        //消息回调
        //开启强制回调
        rabbitTemplate.setMandatory(true);
        //设置消息回调
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int i, String s, String s1, String s2) {
                log.error("消息->{} 路由失败",message);
            }
        });
        return rabbitTemplate;
    }
    @Bean
    public MessageConverter converter(){
        return new Jackson2JsonMessageConverter();
    }
    @Bean
    public DirectExchange delayExchange(RabbitAdmin rabbitAdmin){
        DirectExchange exchange = new DirectExchange(DELAY_EXCHANGE);
        rabbitAdmin.declareExchange(exchange);
        return exchange;
    }
    @Bean
    public DirectExchange dlxExchange(RabbitAdmin rabbitAdmin){
        DirectExchange exchange = new DirectExchange(DLX_EXCHANGE);
        rabbitAdmin.declareExchange(exchange);
        return exchange;
    }
    @Bean
    public Queue delayQueue(RabbitAdmin rabbitAdmin){
        HashMap<String, Object> args = new HashMap<>();
        //死信
        args.put("x-dead-letter-exchange",DLX_EXCHANGE);
        //死信queue
        args.put("x-dead-letter-routing-key","dlx_key");
        //队列的ttl，队列中所有消息的过期时间
        args.put("x-message-ttl",10000);
        //持久化，xx，自动删除，参数
        Queue queue = new Queue(DELAY_QUEUE,true,false,false,args);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }
    @Bean
    public Queue dlxQueue(RabbitAdmin rabbitAdmin){
        Queue queue = new Queue(DLX_QUEUE);
        rabbitAdmin.declareQueue(queue);
        return queue;
    }
    @Bean
    public Binding delayBind(RabbitAdmin rabbitAdmin){
        Binding delay_key = BindingBuilder
                .bind(delayQueue(rabbitAdmin))
                .to(delayExchange(rabbitAdmin))
                .with("delay_key");
        rabbitAdmin.declareBinding(delay_key);
        return delay_key;
    }
    @Bean
    public Binding dlxBind(RabbitAdmin rabbitAdmin){
        Binding dlx_key = BindingBuilder
                .bind(dlxQueue(rabbitAdmin))
                .to(dlxExchange(rabbitAdmin))
                .with("dlx_key");
        rabbitAdmin.declareBinding(dlx_key);
        return dlx_key;
    }
}
//生产
@Component
public class DelaySendMessage {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    private final static String DELAY_EXCHANGE = "delay_exchange";
    private final static String DELAY_QUEUE = "delay_queue";
    private final static String DLX_EXCHANGE = "dlx_exchange";
    private final static String DLX_QUEUE = "dlx_queue";
    public void sendMsg() {
        Message message = MessageBuilder.withBody(
                JSONObject.toJSONString(MessageModel.builder().id(1).context("提现成功!!").build()).getBytes()).build();
        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);
        CorrelationData correlationData = new CorrelationData(String.valueOf(1));
        rabbitTemplate.convertAndSend(DELAY_EXCHANGE,"delay_key",message,correlationData);
    }
}
//消费
@Component
public class ReceiveMessage {
    private final static String DELAY_EXCHANGE = "delay_exchange";
    private final static String DELAY_QUEUE = "delay_queue";
    private final static String DLX_EXCHANGE = "dlx_exchange";
    private final static String DLX_QUEUE = "dlx_queue";
    //延时队列
    @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = DLX_EXCHANGE, type = ExchangeTypes.DIRECT),
            key = "delay_key",
            value = @Queue(value = DLX_QUEUE, autoDelete = "false"),
            ignoreDeclarationExceptions = "true"),
            concurrency = "1", // 指定监听该队列的消费者个数
            ackMode = "MANUAL"// 手动ack
    )
    public void receiveCode(Channel channel, Message msg, @Headers Map<String, Object> headers) throws IOException, InterruptedException {
        log.info(msg.toString());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148