文章目录
写在前面
这是Spring AI Agent系列的第四篇。前面搭了CRUD框架、MCP协议接入、Redis智能缓存。今天解决另一个高频痛点：消息队列的运维噩梦。
用过RabbitMQ的都知道，最烦的不是搭建，是日常运维。消息积压了谁处理？死信队列满了谁清理？消费者挂了谁重启？大部分团队靠告警+人工处理，半夜被叫起来处理消息积压是家常便饭。
我们试了一个方案：把RabbitMQ的管理操作包装成MCP Tool，让AI Agent盯着队列状态，发现问题自动处理。上线一个月，运维告警从每天50条降到了3条。
环境：Spring Boot 3.3.0 + RabbitMQ 3.13 + MCP协议。
一、先看AI Agent能帮你干什么
不用人工介入的场景：
某个队列消息积压超过1000条 → Agent自动扩容消费者
死信队列有消息且原因是业务异常 → Agent自动解析消息内容，尝试修复后重新投递
消费者连续失败3次 → Agent自动暂停该消费者，切换到备用通道
仍需要人工的场景：
消息内容涉及金钱交易（需要人工确认退款金额）
死信原因是数据不存在（可能是数据被误删，需人工恢复）
二、搭建RabbitMQ + Spring Boot基础
pom.xml：
xml
org.springframework.boot
spring-boot-starter-amqp
application.yml：
yaml
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
cache:
channel:
size: 25
publisher-confirm-type: correlated
publisher-returns: true
消息实体和基础配置：
java
@Configuration
public class RabbitMQConfig {
@Bean public TopicExchange orderExchange() { return new TopicExchange("order.exchange"); } @Bean public TopicExchange deadLetterExchange() { return new TopicExchange("dead.letter.exchange"); } @Bean public Queue orderQueue() { return QueueBuilder.durable("order.queue") .deadLetterExchange("dead.letter.exchange") .deadLetterRoutingKey("dead.order") .ttl(30000) .maxLength(10000) .build(); } @Bean public Queue deadLetterQueue() { return QueueBuilder.durable("dead.letter.queue").build(); } @Bean public Binding orderBinding() { return BindingBuilder.bind(orderQueue()) .to(orderExchange()).with("order.*"); } @Bean public Binding deadLetterBinding() { return BindingBuilder.bind(deadLetterQueue()) .to(deadLetterExchange()).with("dead.*"); }}
三、设计消息队列监控数据结构
java
@Data
public class QueueStats {
private String queueName;
private int messageCount;
private int consumerCount;
private double consumeRate;
private long unackedCount;
private List deadLetters;
}
@Data
public class DeadLetterInfo {
private String originalQueue;
private String routingKey;
private String reason;
private String messageBody;
private Date deadTime;
private int retryCount;
}
监控收集Service：
java
@Service
public class QueueMonitorService {
private final RabbitTemplate rabbitTemplate; private final RabbitAdmin rabbitAdmin; public QueueStats getQueueStats(String queueName) { QueueStats stats = new QueueStats(); stats.setQueueName(queueName); AMQP.Queue.DeclareOk declareOk = rabbitAdmin.getRabbitTemplate() .execute(channel -> channel.queueDeclarePassive(queueName)); stats.setMessageCount(declareOk.getMessageCount()); stats.setConsumerCount(declareOk.getConsumerCount()); stats.setConsumeRate(getConsumeRate(queueName)); stats.setUnackedCount(getUnackedCount(queueName)); return stats; } public List<DeadLetterInfo> getDeadLetters(String deadLetterQueue, int limit) { List<DeadLetterInfo> result = new ArrayList<>(); for (int i = 0; i < limit; i++) { Message message = rabbitTemplate.receive(deadLetterQueue, 1000); if (message == null) break; DeadLetterInfo info = parseDeadLetter(message); result.add(info); // 看完放回去 rabbitTemplate.send(deadLetterQueue, message); } return result; } private DeadLetterInfo parseDeadLetter(Message message) { DeadLetterInfo info = new DeadLetterInfo(); Map<String, Object> headers = message.getMessageProperties().getHeaders(); List<Map<String, Object>> deaths = (List<Map<String, Object>>) headers.get("x-death"); if (deaths != null && !deaths.isEmpty()) { Map<String, Object> death = deaths.get(0); info.setReason((String) death.get("reason")); info.setOriginalQueue((String) death.get("queue")); } info.setMessageBody(new String(message.getBody())); info.setDeadTime(message.getMessageProperties().getTimestamp()); Object retryCount = headers.get("x-retry-count"); info.setRetryCount(retryCount != null ? (Integer) retryCount : 0); return info; }}
四、核心：智能消息重试机制
消息消费失败时，不是无脑重试，而是根据失败原因决定策略：
java
@Component
public class SmartRetryHandler {
private final RabbitTemplate rabbitTemplate; public void handleFailure(Message message, Exception cause) { int retryCount = getRetryCount(message); String failureType = classifyFailure(cause); switch (failureType) { case "TEMPORARY": // 临时故障（网络超时、连接池满），延迟重试 if (retryCount < 3) { scheduleRetry(message, retryCount + 1, 5000 * (retryCount + 1)); } else { sendToManualReview(message, cause); } break; case "DATA_ERROR": // 数据问题，尝试自动修复 Message fixed = tryAutoFix(message, cause); if (fixed != null) { rabbitTemplate.send(message.getMessageProperties().getReceivedExchange(), message.getMessageProperties().getReceivedRoutingKey(), fixed); } else { sendToManualReview(message, cause); } break; case "BUSINESS": // 业务异常，直接人工处理 sendToManualReview(message, cause); break; } } private String classifyFailure(Exception cause) { if (cause instanceof TimeoutException || cause instanceof ConnectException) { return "TEMPORARY"; } if (cause instanceof DataIntegrityViolationException) { return "DATA_ERROR"; } return "BUSINESS"; } private void scheduleRetry(Message message, int retryCount, long delayMs) { message.getMessageProperties().setHeader("x-retry-count", retryCount); message.getMessageProperties().setExpiration(String.valueOf(delayMs)); rabbitTemplate.send("retry.exchange", "retry", message); } private void sendToManualReview(Message message, Exception cause) { message.getMessageProperties().setHeader("x-failure-reason", cause.getMessage()); rabbitTemplate.send("manual.review.exchange", "review", message); } private int getRetryCount(Message message) { Object count = message.getMessageProperties().getHeaders().get("x-retry-count"); return count != null ? (Integer) count : 0; } private Message tryAutoFix(Message message, Exception cause) { return null; }}
五、注册为MCP Tool——让AI Agent接管运维
java
@Component
public class QueueManagementTool {
private final QueueMonitorService monitorService; private final RabbitTemplate rabbitTemplate; private final RabbitAdmin rabbitAdmin; @Tool(description = "查询指定队列的实时状态：消息数、消费者数、消费速率。" + "当消息数超过阈值或消费速率异常下降时需要关注") public QueueStats checkQueue( @ToolParam(description = "队列名称，如order.queue") String queueName) { return monitorService.getQueueStats(queueName); } @Tool(description = "查看死信队列中的最近N条消息，分析失败原因。" + "如果发现大量同类型死信，说明存在系统性故障") public List<DeadLetterInfo> inspectDeadLetters( @ToolParam(description = "死信队列名称") String queueName, @ToolParam(description = "查看最近几条，建议10-20条") int limit) { return monitorService.getDeadLetters(queueName, limit); } @Tool(description = "清理死信队列。支持按原因过滤删除。" + "删除前确保已分析原因，避免丢失重要数据") public String purgeDeadLetters( @ToolParam(description = "队列名称") String queueName, @ToolParam(description = "过滤条件：ALL/TEMPORARY/BUSINESS") String reason) { int purged = rabbitAdmin.purgeQueue(queueName, true); return "已清理" + purged + "条死信消息，过滤条件：" + reason; } @Tool(description = "向指定队列发送一条测试消息验证链路是否正常") public String sendTestMessage( @ToolParam(description = "交换机名称") String exchange, @ToolParam(description = "路由键") String routingKey) { String testBody = "{\"type\":\"health_check\",\"timestamp\":" + System.currentTimeMillis() + "}"; rabbitTemplate.convertAndSend(exchange, routingKey, testBody); return "测试消息已发送到 " + exchange + ":" + routingKey; } @Tool(description = "获取全量队列列表和各自的消息积压数") public String listAllQueues() { StringBuilder report = new StringBuilder("队列巡检报告：\n"); String[] queues = {"order.queue", "notification.queue", "dead.letter.queue"}; for (String q : queues) { QueueStats stats = monitorService.getQueueStats(q); String status = stats.getMessageCount() > 5000 ? "⚠️ 积压" : stats.getMessageCount() > 1000 ? "⚡ 注意" : "✅ 正常"; report.append(String.format("%s: %d条消息 %d个消费者 %s\n", q, stats.getMessageCount(), stats.getConsumerCount(), status)); } return report.toString(); }}
六、AI Agent的自治运维流程
每30秒查询所有队列状态。发现 order.queue 消息数超过5000且消费速率下降，自动检查消费者是否存活。消费者挂了→ 告警运维重启。消费者正常但消费慢→ 分析原因并提示扩容。消费者正常但消息量突增→ 临时扩容消费者。
每5分钟扫描死信队列。发现10条以上同类型死信→ 查看消息体分析。临时故障→ 延迟重试。数据错误→ 自动修复后重新投递。业务异常→ 转入人工处理队列附分析报告。
七、踩坑记录
坑1：消息确认和重试的死循环。 消费者处理失败抛异常 → Spring默认重新投递 → 再次失败 → 无限循环。必须在消费者端加重试上限：
java
@RabbitListener(queues = “order.queue”)
public void handleOrder(OrderMessage order, Message message, Channel channel) {
try {
processOrder(order);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
int retryCount = getRetryCount(message);
if (retryCount >= 3) {
channel.basicNack(message.getMessageProperties().getDeliveryTag(),
false, false); // 不重新入队
} else {
channel.basicNack(message.getMessageProperties().getDeliveryTag(),
false, true);
}
}
}
坑2：死信队列无限增长。 消费者挂了一晚上，第二天几十万条死信。一次性清理卡死RabbitMQ，必须分批：
java
int batchSize = 1000;
int totalPurged = 0;
while (true) {
int purged = rabbitAdmin.purgeQueue(“dead.letter.queue”, false);
if (purged == 0) break;
totalPurged += purged;
Thread.sleep(500);
}
坑3：测试消息泛滥。 Agent每30秒发一次测试消息，队列很快全是测试数据。只在消息数或消费者数为0时才发。
八、总结
这套方案把RabbitMQ从"半夜报警把你叫醒"变成了"上班看看Agent的日报"。三个关键设计：智能重试区分故障类型、MCP Tool标准化运维操作、死信自动分析减少人工排查。
如果你也在为消息队列运维头疼，建议先把监控和死信分析这块搭起来——ROI最高的部分。
有用的话点赞收藏，下一篇《AI Agent + Spring Security：MCP协议实现动态权限和自动审计》。
