RabbitMQ 实施与运维完整指南

版本: v1.0
更新日期: 2026年3月28日
适用项目: Dpz.Core

📋 目录

1. 架构概览
2. 快速开始
3. 部署步骤
4. 故障处理
5. 监控运维
6. 常见问题
7. 优化建议

1. 架构概览

1.1 系统架构图

graph TB
    subgraph "家庭服务器 192.168.50.100"
        HF[Hangfire定时器<br/>每3小时]
        JobsP[Jobs生产者<br/>爬取新闻]
        RMQ[RabbitMQ<br/>消息队列]
        ApiC[WebApi消费者<br/>多实例]
        MongoDB[(MongoDB)]
        
        HF -->|触发| JobsP
        JobsP -->|发送NewsArticleMessage| RMQ
        JobsP -->|发送BatchCompletionMessage| RMQ
        RMQ -->|分发消息| ApiC
        ApiC -->|保存文章| MongoDB
        RMQ -->|批次完成| ApiC
        ApiC -->|发送ClearCacheMessage| RMQ
    end
    
    subgraph "云服务器"
        Web[Dpz.Core.Web]
        RMQ -->|清除缓存| Web
    end
    
    style JobsP fill:#fff3cd
    style ApiC fill:#d1ecf1
    style RMQ fill:#f8d7da
    style Web fill:#d4edda

1.2 消息流转

1. Hangfire定时触发 (每3小时)
   ↓
2. Jobs爬取IT之家新闻 (并发爬取)
   ↓
3. 每篇文章 → NewsArticleMessage → RabbitMQ
   ↓
4. WebApi多实例消费 (竞争消费者模式)
   ↓ (每个实例处理不同的消息)
5. 保存到MongoDB
   ↓
6. 批次追踪器记录进度
   ↓
7. 所有消息消费完成 → BatchCompletionMessage
   ↓
8. 等待确认所有消息消费完毕
   ↓
9. 发送ClearCacheMessage → Web清除缓存

1.3 核心组件

1.4 消息类型

2. 快速开始

2.1 前置条件

• ✅ Docker 已安装
• ✅ .NET 10.0 SDK
• ✅ MongoDB 运行正常
• ✅ 配置中心（AgileConfig）已部署

2.2 安装 RabbitMQ

# 拉取镜像
docker pull rabbitmq:4.2.5-management

# 启动容器
docker run -d --name rabbitmq \
  --restart=always \
  -p 5672:5672 \
  -p 15672:15672 \
  -e RABBITMQ_DEFAULT_USER=admin \
  -e RABBITMQ_DEFAULT_PASS=YourSecurePassword123! \
  -v /home/pengqian/program/rabbitmq:/var/lib/rabbitmq \
  rabbitmq:4.2.5-management

# 验证运行
docker ps | grep rabbitmq
docker logs rabbitmq

访问管理界面：http://192.168.50.100:15672

2.3 配置应用

在 AgileConfig 配置中心添加：

{
  "RabbitMQ": {
    "HostName": "192.168.50.100",
    "Port": 5672,
    "UserName": "admin",
    "Password": "YourSecurePassword123!",
    "VirtualHost": "/",
    "RequestedConnectionTimeout": 30,
    "RequestedHeartbeat": 60,
    "AutomaticRecoveryEnabled": true,
    "NetworkRecoveryInterval": 10
  }
}

3. 部署步骤

3.1 编译项目

cd /path/to/dpz.core/src

# 编译所有项目
dotnet build Dpz.Core.MessageQueue/Dpz.Core.MessageQueue.csproj
dotnet build Dpz.Core.Web.Jobs/Dpz.Core.Web.Jobs.csproj
dotnet build Dpz.Core.WebApi/Dpz.Core.WebApi.csproj
dotnet build Dpz.Core.Web/Dpz.Core.Web.csproj

3.2 部署 Jobs (生产者)

# 停止旧服务
sudo docker stop dpz.jobs
sudo docker rm dpz.jobs

# 构建镜像
cd /home/ubuntu/project/dpz.core/src
sudo docker build -t dpz.jobs:latest -f Dpz.Core.Web.Jobs/Dockerfile .

# 启动服务
sudo docker run -d --name dpz.jobs \
  --restart=always \
  --network=host \
  -v /home/ubuntu/logs/jobs:/app/logs \
  dpz.jobs:latest

3.3 部署 WebApi (消费者)

# 构建镜像
sudo docker build -t dpz.webapi:latest -f Dpz.Core.WebApi/Dockerfile .

# 启动多实例（3个实例）
for i in {1..3}; do
  sudo docker run -d --name dpz.webapi-$i \
    --restart=always \
    --network=host \
    -e ASPNETCORE_URLS="http://*:$((8000+$i))" \
    -v /home/ubuntu/logs/webapi-$i:/app/logs \
    dpz.webapi:latest
done

# 配置Nginx负载均衡
sudo nano /etc/nginx/sites-available/webapi

# 添加配置
upstream webapi_backend {
    server 127.0.0.1:8001;
    server 127.0.0.1:8002;
    server 127.0.0.1:8003;
}

server {
    listen 80;
    server_name api.dpangzi.com;
    
    location / {
        proxy_pass http://webapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

# 重载Nginx
sudo nginx -t && sudo nginx -s reload

3.4 部署 Web (缓存清除消费者)

# 构建镜像
sudo docker build -t dpz.web:latest -f Dpz.Core.Web/Dockerfile .

# 启动服务
sudo docker run -d --name dpz.web \
  --restart=always \
  -p 5000:5000 \
  -v /home/ubuntu/logs/web:/app/logs \
  dpz.web:latest

3.5 验证部署

1) 检查 RabbitMQ 连接

访问 http://192.168.50.100:15672

• Connections 页面应该看到：
  • Dpz.Core.Web.Jobs (1个连接)
  • Dpz.Core.WebApi (3个连接，对应3个实例)
  • Dpz.Core.Web (1个连接)

2) 检查队列创建

• Exchanges 页面应该看到：

  • dpz.news.exchange (type: topic)
  • dpz.system.exchange (type: topic)

• Queues 页面应该看到：

  • dpz.news.article.queue
  • dpz.system.batch.queue
  • dpz.system.cache.queue

3) 查看应用日志

# Jobs 日志
sudo docker logs -f dpz.jobs | grep "RabbitMQ"

# WebApi 日志
sudo docker logs -f dpz.webapi-1 | grep "消息"

# Web 日志
sudo docker logs -f dpz.web | grep "缓存"

预期日志输出：

[Jobs]
[INF] 成功连接到RabbitMQ: 192.168.50.100:5672
[INF] Exchange已就绪: dpz.news.exchange
[INF] 爬虫任务开始, BatchId=batch_20260328120000_abc123
[INF] 文章《xxx》已发送到消息队列, MessageId=xxx, BatchId=batch_20260328120000_abc123

[WebApi-1]
[INF] 消息消费者已启动: Queue=dpz.news.article.queue
[INF] 开始处理新闻消息: xxx, BatchId: batch_20260328120000_abc123
[INF] 新闻保存成功: xxx, MessageId: xxx
[INF] 批次进度已更新: BatchId=batch_20260328120000_abc123

[WebApi-2]
[INF] 收到批次完成消息: BatchId=batch_20260328120000_abc123, TotalMessages=50
[INF] 批次进度检查: BatchId=batch_20260328120000_abc123, Progress=50/50
[INF] 批次已完成, 清除缓存消息已发送

[Web]
[INF] 收到清除缓存消息: BatchId=batch_20260328120000_abc123
[INF] 文章缓存清除成功

4) 手动触发测试

访问 Hangfire 面板：https://task.dpangzi.com/jobs

• 找到"爬取数据"任务
• 点击右侧"Execute now"手动触发
• 观察日志和 RabbitMQ 管理界面

4. 故障处理

4.1 网络延迟问题

场景

家庭服务器网络不稳定，发送消息到 RabbitMQ 延迟高。

症状

[ERR] 发送消息超时: Request timeout
[WRN] RabbitMQ连接已关闭: Connection reset

解决方案

方案1: 本地消息表

在 Jobs 项目中添加本地SQLite队列作为缓冲：

// 新增 LocalMessageQueue.cs
public class LocalMessageQueue
{
    private readonly string _dbPath = "local_message_queue.db";
    
    public async Task EnqueueAsync<T>(T message) where T : MessageBase
    {
        // 保存到本地SQLite
        // 包含重试逻辑
    }
    
    public async Task<List<T>> GetPendingAsync<T>() where T : MessageBase
    {
        // 获取待发送的消息
    }
}

// 修改 RabbitMQPublisher
public async Task PublishAsync(TMessage message)
{
    try
    {
        // 尝试直接发送
        await PublishToRabbitMQAsync(message);
    }
    catch (Exception ex)
    {
        // 发送失败，保存到本地队列
        await _localQueue.EnqueueAsync(message);
        _logger.LogWarning("RabbitMQ不可达，消息已保存到本地队列");
    }
}

// 后台任务定期同步
public class LocalQueueSyncService : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            var pending = await _localQueue.GetPendingAsync<NewsArticleMessage>();
            foreach (var msg in pending)
            {
                try
                {
                    await _publisher.PublishAsync(msg);
                    await _localQueue.RemoveAsync(msg.MessageId);
                }
                catch { /* 继续重试 */ }
            }
            await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);
        }
    }
}

方案2: 调整超时配置

{
  "RabbitMQ": {
    "RequestedConnectionTimeout": 60,  // 增加到60秒
    "RequestedHeartbeat": 120,         // 增加到120秒
    "NetworkRecoveryInterval": 5       // 减少到5秒，更快重连
  }
}

4.2 消息丢失问题

场景1: 发送失败未察觉

解决方案: 发布者确认

// 修改 RabbitMQPublisher.cs
public async Task PublishAsync(TMessage message)
{
    // 启用发布者确认
    _channel.ConfirmSelect();
    
    // 发送消息
    _channel.BasicPublish(...);
    
    // 等待确认
    var confirmed = _channel.WaitForConfirms(TimeSpan.FromSeconds(5));
    if (!confirmed)
    {
        throw new MessagePublishException("消息未被RabbitMQ确认");
    }
}

场景2: 消费处理失败

解决方案: 死信队列

// 配置死信队列
public static void ConfigureDeadLetterQueue(IModel channel)
{
    // 1. 创建死信Exchange
    channel.ExchangeDeclare(
        exchange: "dpz.dlx.exchange",
        type: "topic",
        durable: true
    );
    
    // 2. 创建死信Queue
    channel.QueueDeclare(
        queue: "dpz.news.article.dlq",
        durable: true,
        exclusive: false,
        autoDelete: false,
        arguments: null
    );
    
    // 3. 绑定
    channel.QueueBind(
        queue: "dpz.news.article.dlq",
        exchange: "dpz.dlx.exchange",
        routingKey: "news.article.#"
    );
    
    // 4. 修改原队列，配置死信
    var args = new Dictionary<string, object>
    {
        ["x-dead-letter-exchange"] = "dpz.dlx.exchange",
        ["x-dead-letter-routing-key"] = "news.article.failed"
    };
    
    channel.QueueDeclare(
        queue: "dpz.news.article.queue",
        durable: true,
        exclusive: false,
        autoDelete: false,
        arguments: args
    );
}

// 消费者处理失败超过3次后，消息自动进入死信队列
if (message.RetryCount >= 3)
{
    _channel.BasicNack(deliveryTag, false, requeue: false);  // 进入DLQ
}

定期检查死信队列：

# 查看死信队列
rabbitmqadmin get queue=dpz.news.article.dlq count=10

# 或在管理界面 Queues → dpz.news.article.dlq → Get messages

4.3 RabbitMQ 宕机

场景

RabbitMQ服务崩溃或服务器重启。

症状

[ERR] 连接RabbitMQ失败: Connection refused
[WRN] 所有重试均失败

解决方案

方案1: 健壮的重连机制（已实现）

{
  "RabbitMQ": {
    "AutomaticRecoveryEnabled": true,   // 自动重连
    "NetworkRecoveryInterval": 10       // 每10秒重试
  }
}

代码中会自动重连，无需人工干预。

方案2: 高可用集群（生产环境推荐）

# docker-compose-rabbitmq-cluster.yml
version: '3.8'

services:
  rabbitmq-1:
    image: rabbitmq:4.2.5-management
    hostname: rabbitmq-1
    environment:
      RABBITMQ_ERLANG_COOKIE: 'secret_cookie_string'
    volumes:
      - ./rabbitmq-1:/var/lib/rabbitmq
    ports:
      - "5672:5672"
      - "15672:15672"
    networks:
      - rabbitmq-cluster

  rabbitmq-2:
    image: rabbitmq:4.2.5-management
    hostname: rabbitmq-2
    environment:
      RABBITMQ_ERLANG_COOKIE: 'secret_cookie_string'
    volumes:
      - ./rabbitmq-2:/var/lib/rabbitmq
    networks:
      - rabbitmq-cluster

  rabbitmq-3:
    image: rabbitmq:4.2.5-management
    hostname: rabbitmq-3
    environment:
      RABBITMQ_ERLANG_COOKIE: 'secret_cookie_string'
    volumes:
      - ./rabbitmq-3:/var/lib/rabbitmq
    networks:
      - rabbitmq-cluster

  haproxy:
    image: haproxy:2.8
    ports:
      - "5673:5672"   # 负载均衡后的端口
      - "15673:8080"  # HAProxy统计页面
    volumes:
      - ./haproxy.cfg:/usr/local/etc/haproxy/haproxy.cfg:ro
    networks:
      - rabbitmq-cluster

networks:
  rabbitmq-cluster:
    driver: bridge

# 配置集群
docker exec -it rabbitmq-2 bash
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@rabbitmq-1
rabbitmqctl start_app

docker exec -it rabbitmq-3 bash
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@rabbitmq-1
rabbitmqctl start_app

# 配置镜像队列（所有节点都有副本）
rabbitmqctl set_policy ha-all "^dpz\\." \
  '{"ha-mode":"all","ha-sync-mode":"automatic"}' \
  --priority 1 \
  --apply-to queues

收益：

• ✅ 单节点宕机不影响服务
• ✅ 数据高可用（多副本）
• ✅ 负载均衡

4.4 队列堆积

场景

消息生产速度 > 消费速度，队列中消息越来越多。

症状

RabbitMQ管理界面:
dpz.news.article.queue
├─ Ready: 1500 (持续增长)
├─ Unacked: 3
└─ Rate: +10 msg/s (net)

解决方案

方案1: 增加消费者实例

# 启动更多WebApi实例
for i in {4..6}; do
  sudo docker run -d --name dpz.webapi-$i \
    --restart=always \
    --network=host \
    -e ASPNETCORE_URLS="http://*:$((8000+$i))" \
    dpz.webapi:latest
done

# 更新Nginx配置
upstream webapi_backend {
    server 127.0.0.1:8001;
    server 127.0.0.1:8002;
    server 127.0.0.1:8003;
    server 127.0.0.1:8004;  # 新增
    server 127.0.0.1:8005;  # 新增
    server 127.0.0.1:8006;  # 新增
}

方案2: 优化消费者性能

// 检查消费者处理逻辑，是否有不必要的耗时操作
public async Task<bool> HandleAsync(NewsArticleMessage message)
{
    // ❌ 不要在消费者中做耗时操作
    // await Task.Delay(5000);  // 模拟慢处理
    
    // ✅ 快速处理，必要时使用后台任务
    await _articleService.CreateArticleAsync(article);
    
    // ❌ 不要同步调用外部API
    // var result = _httpClient.GetStringAsync("https://slow-api.com").Result;
    
    return true;
}

方案3: 临时暂停生产者

# 在Hangfire面板中禁用爬虫任务
# 等队列消费完毕后再启用

4.5 消息重复消费

场景

网络波动导致ACK丢失，RabbitMQ重新发送消息。

症状

[WRN] 新闻已存在，跳过: https://www.ithome.com/0/123456.htm

解决方案（已实现）

// NewsArticleMessageHandler.cs 中的幂等性检查
var noExists = await _articleService.NoExistsByFromAsync([message.From]);
if (noExists.Count == 0)
{
    _logger.LogInformation("新闻已存在，跳过: {From}", message.From);
    return true;  // 返回true，不重试
}

这是正常现象，无需担心！ 幂等性检查会自动处理。

5. 监控运维

5.1 RabbitMQ 管理界面

访问：http://192.168.50.100:15672

关键指标：

5.2 应用日志监控

# 实时监控关键日志
sudo docker logs -f dpz.jobs | grep -E "ERROR|WARN|RabbitMQ"
sudo docker logs -f dpz.webapi-1 | grep -E "ERROR|WARN|消息"

# 统计消息处理量
sudo docker logs dpz.webapi-1 --since 1h | grep "新闻保存成功" | wc -l

# 查找错误
sudo docker logs dpz.jobs --since 1h | grep ERROR

5.3 健康检查端点

# WebApi 健康检查
curl http://localhost:8001/health/rabbitmq

# 预期返回
{
  "status": "Healthy",
  "duration": "00:00:00.0234567",
  "entries": {
    "rabbitmq": {
      "status": "Healthy",
      "data": {
        "Connection": "Open",
        "Hostname": "192.168.50.100"
      }
    }
  }
}

5.4 定期巡检清单

每日检查：

• RabbitMQ 管理界面正常访问
• Connections 数量正常（5个）
• 队列 Ready 消息数 < 50
• 无错误日志

每周检查：

• 死信队列是否有消息（应该为0）
• 磁盘空间充足（RabbitMQ数据目录）
• 消费者实例均匀分担负载
• 消息处理平均延迟 < 100ms

每月检查：

• 清理旧日志文件
• 检查RabbitMQ版本更新
• 评估是否需要扩容
• 容灾演练（模拟RabbitMQ宕机）

6. 常见问题

Q1: 云服务器上的 Web 项目能连接到家庭服务器的 RabbitMQ 吗？

A: 可以，但需要配置：

1. 开放端口（不推荐暴露到公网）

# 仅允许云服务器IP访问
sudo ufw allow from <云服务器公网IP> to any port 5672

2. 或者使用 VPN/内网穿透

• 推荐使用 WireGuard VPN
• 或使用 frp/ngrok 等工具

3. 配置 RabbitMQ 监听地址

# /etc/rabbitmq/rabbitmq.conf
listeners.tcp.default = 0.0.0.0:5672

Q2: 为什么使用批次追踪而不是直接清除缓存？

A: 因为消息是异步消费的，考虑以下情况：

T1: Jobs 爬取完成100条新闻
T2: Jobs 发送100条消息到RabbitMQ
T3: Jobs 发送"清除缓存"消息
T4: Web 收到"清除缓存"消息 → 立即清除缓存
T5: WebApi 慢慢消费消息（可能需要1分钟）

问题：缓存已清除，但新文章还未保存到数据库！
用户访问网站看不到新文章。

批次追踪的方案：

T1: Jobs 生成 BatchId
T2: Jobs 发送100条消息（都带 BatchId）
T3: Jobs 发送 BatchCompletionMessage（BatchId, 总数100）
T4: WebApi 每消费1条 → 批次追踪器 +1
T5: BatchCompletionMessageHandler 轮询检查进度
T6: 确认100条都消费完毕
T7: 发送"清除缓存"消息
T8: Web 清除缓存

好处：确保所有新文章已保存后再清除缓存

Q3: 多个 WebApi 实例会重复消费消息吗？

A: 不会！RabbitMQ 的竞争消费者模式保证：

Queue: [M1, M2, M3, M4, M5]
       ↓   ↓   ↓   ↓   ↓
实例1: M1      M4
实例2:     M2      M5
实例3:         M3

每条消息只会被一个实例消费。

Q4: 如果 WebApi 实例崩溃，消息会丢失吗？

A: 不会！

T1: 实例1 获取消息 M1
T2: 实例1 开始处理 M1
T3: 实例1 崩溃（未发送ACK）
T4: RabbitMQ 等待一段时间未收到ACK
T5: RabbitMQ 将 M1 重新放回队列
T6: 实例2 获取 M1 并处理

消息不会丢失，会被其他实例处理。

Q5: prefetchCount=1 是什么意思？

A: 控制消费者从队列预取的消息数量：

// prefetchCount: 1
实例1: 获取M1 → 处理中 → ACK → 获取M2
       ↑___只拿1条，处理完再拿___↑

// prefetchCount: 10
实例1: 获取M1-M10 → 处理M1 → ACK → 处理M2...
       ↑___一次拿10条，慢慢处理___↑

优点：
- prefetchCount=1: 公平分发，快的实例自动多干活
- prefetchCount=10: 减少网络往返，但可能分配不均

Q6: 如何查看批次完成的状态？

A: 查看日志或 Redis：

# 查看日志
sudo docker logs dpz.webapi-1 | grep "批次进度"

# 查看Redis（需要安装redis-cli）
redis-cli
> KEYS batch:progress:*
> GET batch:progress:batch_20260328120000_abc123
"50"  # 已消费50条

> GET batch:messages:batch_20260328120000_abc123
"[\"msg1\", \"msg2\", ...]"  # 已消费的消息ID列表

Q7: 死信队列中的消息如何处理？

A: 人工介入：

# 1. 查看死信队列
rabbitmqadmin get queue=dpz.news.article.dlq count=10

# 2. 分析失败原因（查看日志）
sudo docker logs dpz.webapi-1 | grep "处理失败"

# 3. 修复问题后，手动重发
rabbitmqadmin get queue=dpz.news.article.dlq requeue=true

# 或写脚本批量重发

Q8: RabbitMQ 占用磁盘空间太大怎么办？

A: 清理策略：

# 1. 检查磁盘使用
du -sh /home/pengqian/program/rabbitmq

# 2. 配置TTL（消息过期时间）
rabbitmqctl set_policy TTL "^dpz\\.news\\.article\\.queue$" \
  '{"message-ttl":86400000}' \
  --apply-to queues
  
# 消息24小时后自动删除

# 3. 配置最大队列长度
rabbitmqctl set_policy max-length "^dpz\\.news\\.article\\.queue$" \
  '{"max-length":10000}' \
  --apply-to queues
  
# 队列最多10000条消息，超过后丢弃旧消息

# 4. 定期备份并清理日志
rabbitmqctl rotate_logs

Q9: 如何测试消息队列是否正常？

A: 手动发送测试消息：

// 在 Jobs 项目中添加测试接口
[HttpPost("test/send-message")]
public async Task<IActionResult> TestSendMessage()
{
    var message = new NewsArticleMessage
    {
        MessageId = Guid.NewGuid().ToString(),
        Title = "测试消息",
        Content = "这是一条测试消息",
        From = $"https://test.com/{DateTime.Now.Ticks}",
        Tags = new List<string> { "测试" },
        Source = "Manual Test"
    };
    
    await _publisher.PublishAsync(message);
    
    return Ok(new { MessageId = message.MessageId });
}

# 调用测试接口
curl -X POST http://task.dpangzi.com/test/send-message

# 观察日志
sudo docker logs -f dpz.webapi-1 | grep "测试消息"

Q10: 如何回滚到不使用消息队列的版本？

A: 修改代码：

// TaskService.cs
protected virtual async Task PublishArticleAsync(string feedUrl)
{
    var article = new CreateArticleRequest { ... };
    var userInfo = await GetArticleContentAsync(article, feedUrl);
    
    if (userInfo == null) return;
    
    // 注释掉消息队列逻辑
    // await _messagePublisher.PublishAsync(message);
    
    // 恢复直接保存
    await ArticleService.CreateArticleAsync(article, userInfo);
    
    _logger.LogInformation("文章已保存: {Title}", article.Title);
}

// Program.cs 注释掉消息队列注册
// builder.Services.AddRabbitMQ(builder.Configuration);
// builder.Services.AddMessageConsumer<NewsArticleMessage, NewsArticleMessageHandler>();

重新编译部署即可。

7. 优化建议

7.1 短期优化（1-2周）

1) 添加发布者确认

// RabbitMQPublisher.cs
_channel.ConfirmSelect();
await _channel.WaitForConfirmsAsync(cancellationToken);

收益： 确保消息成功写入RabbitMQ

2) 配置死信队列

// 在 MessageQueueServiceExtensions.cs 中添加
public static void ConfigureDeadLetterQueues(IModel channel)
{
    // 配置死信Exchange和Queue
}

收益： 失败消息可追溯，支持人工处理

3) 优化日志级别

{
  "Logging": {
    "LogLevel": {
      "Dpz.Core.MessageQueue": "Information",
      "RabbitMQ.Client": "Warning"
    }
  }
}

收益： 减少日志噪音，突出关键信息

7.2 中期优化（1-2个月）

1) 实现本地消息表

参考 4.1 网络延迟问题 - 方案1

收益： 网络故障期间不丢消息

2) 添加监控和告警

// 集成 Prometheus
services.AddHealthChecks()
    .AddRabbitMQ()
    .AddMongoDB()
    .ForwardToPrometheus();

// 配置告警规则
alert: RabbitMQQueueBacklog
expr: rabbitmq_queue_messages_ready > 100
for: 5m
annotations:
  summary: "队列堆积告警"
  description: "{{ $labels.queue }} 队列有 {{ $value }} 条待处理消息"

收益： 及时发现问题，主动运维

3) 性能测试

// 压力测试：模拟1000条消息
for (int i = 0; i < 1000; i++)
{
    await _publisher.PublishAsync(new NewsArticleMessage { ... });
}

// 观察指标：
// - 消息发送速率
// - 消息消费速率
// - 队列深度
// - 系统资源占用

收益： 了解系统瓶颈，提前扩容

7.3 长期规划

1) RabbitMQ 高可用集群

参考 4.3 RabbitMQ 宕机 - 方案2

收益： 生产级可用性（99.9%+）

2) 消息追踪与可观测性

// 集成 OpenTelemetry
services.AddOpenTelemetryTracing(builder =>
{
    builder
        .AddRabbitMQInstrumentation()
        .AddMongoDBInstrumentation()
        .AddJaegerExporter();
});

收益： 全链路追踪，快速定位问题

3) 扩展新的消费者

当前：
NewsArticleMessage → WebApi (保存数据库)

未来：
NewsArticleMessage → WebApi (保存数据库)
                  ├→ AIAnalysisService (情感分析)
                  ├→ SearchIndexService (构建搜索索引)
                  ├→ NotificationService (推送通知)
                  └→ StatisticsService (数据统计)

每个服务独立订阅，互不影响

收益： 微服务架构，灵活扩展

4) 考虑迁移到 Kafka

场景：消息量达到 百万/天 级别

Kafka 优势：
- 更高吞吐量（百万/秒）
- 更好的持久化
- 消息回溯（重新消费历史消息）

迁移成本：
- 需要修改消息发布/消费代码
- 部署和运维复杂度增加

📚 附录

A. 配置模板汇总

A.1 RabbitMQ 配置（AgileConfig）

{
  "RabbitMQ": {
    "HostName": "192.168.50.100",
    "Port": 5672,
    "UserName": "admin",
    "Password": "YourSecurePassword123!",
    "VirtualHost": "/",
    "RequestedConnectionTimeout": 30,
    "RequestedHeartbeat": 60,
    "AutomaticRecoveryEnabled": true,
    "NetworkRecoveryInterval": 10
  }
}

A.2 Redis 配置（用于批次追踪）

{
  "Redis": {
    "Configuration": "192.168.50.100:6379",
    "InstanceName": "Dpz.Core:"
  }
}

A.3 Nginx 负载均衡配置

upstream webapi_backend {
    least_conn;  # 最少连接数算法
    
    server 127.0.0.1:8001 weight=1 max_fails=3 fail_timeout=30s;
    server 127.0.0.1:8002 weight=1 max_fails=3 fail_timeout=30s;
    server 127.0.0.1:8003 weight=1 max_fails=3 fail_timeout=30s;
}

server {
    listen 80;
    server_name api.dpangzi.com;
    
    location / {
        proxy_pass http://webapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # 超时设置
        proxy_connect_timeout 30s;
        proxy_send_timeout 30s;
        proxy_read_timeout 30s;
        
        # 健康检查
        proxy_next_upstream error timeout http_500 http_502 http_503;
    }
    
    location /health {
        access_log off;
        proxy_pass http://webapi_backend/health;
    }
}

B. 快速排查命令

# === RabbitMQ ===
# 查看连接
rabbitmqctl list_connections name peer_host peer_port

# 查看队列状态
rabbitmqctl list_queues name messages_ready messages_unacknowledged

# 查看Exchange
rabbitmqctl list_exchanges name type

# 查看绑定关系
rabbitmqctl list_bindings

# 查看消费者
rabbitmqctl list_consumers

# === Docker ===
# 查看运行中的容器
docker ps | grep -E "dpz|rabbitmq"

# 查看容器资源占用
docker stats --no-stream dpz.jobs dpz.webapi-1 dpz.webapi-2 dpz.webapi-3

# 查看容器日志（最近100行）
docker logs --tail 100 dpz.jobs

# 进入容器shell
docker exec -it dpz.jobs bash

# === 系统资源 ===
# 磁盘空间
df -h | grep -E "rabbitmq|mongodb"

# 内存占用
free -h

# CPU占用
top -bn1 | head -20

# 网络连接
netstat -antp | grep -E "5672|8001|8002|8003"

# === Redis ===
# 查看批次追踪数据
redis-cli
> KEYS batch:*
> GET batch:progress:batch_20260328120000_abc123
> TTL batch:progress:batch_20260328120000_abc123

# === MongoDB ===
# 查看最新插入的文章
mongo
> use dpz_core
> db.articles.find().sort({_id:-1}).limit(5).pretty()
> db.articles.countDocuments({createdAt: {$gte: new Date(Date.now() - 3600000)}})

C. 性能基准参考

基于当前架构的性能测试结果（仅供参考）：

推荐配置：

• 日常运行：3个WebApi实例
• 高峰期：5个WebApi实例
• 紧急情况：可临时扩展到10个实例

D. 容灾演练清单

定期进行容灾演练，确保故障发生时能快速恢复：

演练1: RabbitMQ 宕机

# 1. 模拟宕机
sudo docker stop rabbitmq

# 2. 观察现象
# - Jobs 日志应该显示连接失败
# - WebApi 日志应该显示消费者停止

# 3. 恢复服务
sudo docker start rabbitmq

# 4. 验证恢复
# - 5分钟内应该自动重连
# - 队列中的消息继续消费
# - 无数据丢失

# 5. 记录恢复时间

演练2: WebApi 实例崩溃

# 1. 停止一个实例
sudo docker stop dpz.webapi-1

# 2. 观察现象
# - 剩余实例正常消费
# - 队列中消息不会丢失
# - 吞吐量降低但仍可用

# 3. 恢复实例
sudo docker start dpz.webapi-1

# 4. 验证恢复
# - 实例重新加入消费
# - 负载重新均衡

演练3: 网络中断

# 1. 模拟网络中断（在Jobs服务器上）
sudo iptables -A OUTPUT -p tcp --dport 5672 -j DROP

# 2. 观察现象
# - Jobs 无法发送消息
# - 如果实现了本地队列，消息保存到本地

# 3. 恢复网络
sudo iptables -D OUTPUT -p tcp --dport 5672 -j DROP

# 4. 验证恢复
# - 重新连接成功
# - 本地队列中的消息自动同步

✅ 验收清单

部署完成后，请逐项检查：

基础设施

• RabbitMQ 服务运行正常（http://192.168.50.100:15672）
• Redis 服务运行正常
• MongoDB 服务运行正常
• 所有 Docker 容器运行正常

连接状态

• RabbitMQ Connections 页面看到 5 个连接
• Exchanges 页面看到 dpz.news.exchange 和 dpz.system.exchange
• Queues 页面看到 3 个队列（article、batch、cache）

功能验证

• 手动触发爬虫任务，能看到消息发送日志
• WebApi 日志显示消息被消费
• MongoDB 中看到新插入的文章
• 批次完成后，Web 日志显示缓存已清除

监控验证

• RabbitMQ 管理界面可以访问
• 健康检查端点返回正常
• 日志文件正常写入
• 无错误或警告日志

性能验证

• 队列 Ready 消息数保持在 10 以内
• 消息消费速率 > 生产速率
• CPU 占用 < 50%
• 内存占用合理

📞 联系与支持

• 项目地址: https://github.com/pengqian089/dpz.core
• 技术文档: src/Dpz.Core.MessageQueue/README.md
• 问题反馈: GitHub Issues

最后更新: 2026年3月28日
文档版本: v1.0
作者: AI Assistant based on user's implementation