feat(rabbitmq): 新增死信队列(DLQ)支持

当前 RabbitMQ 已实现 Confirm + Returns + Retry,但缺少 DLQ 导致重试耗尽后消息直接丢失,无法追溯和人工补偿。

新增:
- RabbitMqDeadLetterConfig:声明 DLX 交换机及 send/recall 死信队列
- RabbitMqDeadLetterReceiver:消费死信消息,记录 headers 与 body 日志
- 主队列绑定 x-dead-letter-exchange / x-dead-letter-routing-key
- 配置 default-requeue-rejected=false,确保重试耗尽后进入 DLQ 而非 requeue

消息流向:
austin.queues.send/recall → 消费失败 → 重试 3 次 → 耗尽 →
austin.dlx → austin.queues.send.dead/recall.dead → 日志告警
pull/89/head
Rangsh 2 weeks ago
parent 8379c029e0
commit cd398b40ee

@ -0,0 +1,38 @@
package com.java3y.austin.handler.receiver.rabbit;
import com.java3y.austin.support.constans.MessageQueuePipeline;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
/**
*
* DLQ 便
*
* @author Rangsh
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "austin.mq.pipeline", havingValue = MessageQueuePipeline.RABBIT_MQ)
public class RabbitMqDeadLetterReceiver {
@RabbitListener(queues = "${austin.rabbitmq.queues.send.dead}")
public void onSendDeadLetter(Message message) {
log.error("[DLQ][send] headers={}, body={}",
message.getMessageProperties().getHeaders(),
new String(message.getBody(), StandardCharsets.UTF_8));
// TODO: 后续可扩展落库 / 告警 / 人工补偿入口
}
@RabbitListener(queues = "${austin.rabbitmq.queues.recall.dead}")
public void onRecallDeadLetter(Message message) {
log.error("[DLQ][recall] headers={}, body={}",
message.getMessageProperties().getHeaders(),
new String(message.getBody(), StandardCharsets.UTF_8));
// TODO: 后续可扩展落库 / 告警 / 人工补偿入口
}
}

@ -9,6 +9,7 @@ import com.java3y.austin.support.constans.MessageQueuePipeline;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Argument;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
@ -33,7 +34,14 @@ public class RabbitMqReceiver implements MessageReceiver {
private ConsumeService consumeService;
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "${spring.rabbitmq.queues.send}", durable = "true"),
value = @Queue(
value = "${spring.rabbitmq.queues.send}",
durable = "true",
arguments = {
@Argument(name = "x-dead-letter-exchange", value = "${austin.rabbitmq.dlx.exchange}"),
@Argument(name = "x-dead-letter-routing-key", value = "${austin.rabbitmq.routing.send.dead}")
}
),
exchange = @Exchange(value = "${austin.rabbitmq.exchange.name}", type = ExchangeTypes.TOPIC),
key = "${austin.rabbitmq.routing.send}"
))
@ -49,7 +57,14 @@ public class RabbitMqReceiver implements MessageReceiver {
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "${spring.rabbitmq.queues.recall}", durable = "true"),
value = @Queue(
value = "${spring.rabbitmq.queues.recall}",
durable = "true",
arguments = {
@Argument(name = "x-dead-letter-exchange", value = "${austin.rabbitmq.dlx.exchange}"),
@Argument(name = "x-dead-letter-routing-key", value = "${austin.rabbitmq.routing.recall.dead}")
}
),
exchange = @Exchange(value = "${austin.rabbitmq.exchange.name}", type = ExchangeTypes.TOPIC),
key = "${austin.rabbitmq.routing.recall}"
))

@ -0,0 +1,66 @@
package com.java3y.austin.support.config;
import com.java3y.austin.support.constans.MessageQueuePipeline;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* RabbitMQ
* DLQ便
*
* @author Rangsh
*/
@Configuration
@ConditionalOnProperty(name = "austin.mq.pipeline", havingValue = MessageQueuePipeline.RABBIT_MQ)
public class RabbitMqDeadLetterConfig {
@Value("${austin.rabbitmq.dlx.exchange}")
private String dlxExchangeName;
@Value("${austin.rabbitmq.queues.send.dead}")
private String sendDeadQueueName;
@Value("${austin.rabbitmq.queues.recall.dead}")
private String recallDeadQueueName;
@Value("${austin.rabbitmq.routing.send.dead}")
private String sendDeadRoutingKey;
@Value("${austin.rabbitmq.routing.recall.dead}")
private String recallDeadRoutingKey;
@Bean
public DirectExchange deadLetterExchange() {
return new DirectExchange(dlxExchangeName, true, false);
}
@Bean
public Queue sendDeadLetterQueue() {
return new Queue(sendDeadQueueName, true);
}
@Bean
public Queue recallDeadLetterQueue() {
return new Queue(recallDeadQueueName, true);
}
@Bean
public Binding sendDeadLetterBinding() {
return BindingBuilder.bind(sendDeadLetterQueue())
.to(deadLetterExchange())
.with(sendDeadRoutingKey);
}
@Bean
public Binding recallDeadLetterBinding() {
return BindingBuilder.bind(recallDeadLetterQueue())
.to(deadLetterExchange())
.with(recallDeadRoutingKey);
}
}

@ -71,11 +71,19 @@ spring.rabbitmq.listener.simple.retry.enabled=true
spring.rabbitmq.listener.simple.retry.initial-interval=1000
spring.rabbitmq.listener.simple.retry.multiplier=2
spring.rabbitmq.listener.simple.retry.max-attempts=3
# retry exhausted → reject (no requeue) → DLQ
spring.rabbitmq.listener.simple.default-requeue-rejected=false
austin.rabbitmq.exchange.name=austin.point
spring.rabbitmq.queues.send=austin.queues.send
spring.rabbitmq.queues.recall=austin.queues.recall
austin.rabbitmq.routing.send=austin.send
austin.rabbitmq.routing.recall=austin.recall
# dead letter
austin.rabbitmq.dlx.exchange=austin.dlx
austin.rabbitmq.queues.send.dead=austin.queues.send.dead
austin.rabbitmq.queues.recall.dead=austin.queues.recall.dead
austin.rabbitmq.routing.send.dead=austin.send.dead
austin.rabbitmq.routing.recall.dead=austin.recall.dead
########################################## RabbitMq end ##########################################

Loading…
Cancel
Save