写在前面
这是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协议实现动态权限和自动审计》。