RabbitMQ 运维实战：集群部署、消费者可靠性与监控体系

# 在所有节点执行，确保主机名互相解析
cat >> /etc/hosts << EOF
10.0.1.11 mq-node1
10.0.1.12 mq-node2
10.0.1.13 mq-node3
EOF

# 统一 Erlang Cookie（所有节点相同）
echo "RABBITMQ_ERLANG_COOKIE_STRING" > /var/lib/rabbitmq/.erlang.cookie
chown rabbitmq:rabbitmq /var/lib/rabbitmq/.erlang.cookie
chmod 400 /var/lib/rabbitmq/.erlang.cookie

# /etc/rabbitmq/rabbitmq.conf

# 节点名称（每个节点不同）
# node 名称通过环境变量 RABBITMQ_NODENAME 或 systemd 设置

# 监听配置
listeners.tcp.default = 5672
management.tcp.port = 15672

# 集群分区策略（pause_minority 是 Quorum Queue 的推荐策略）
cluster_partition_handling = pause_minority

# 日志级别
log.console = true
log.console.level = info
log.file = /var/log/rabbitmq/rabbit.log
log.file.level = info

# 停止 RabbitMQ 应用（Erlang 节点保持运行）
rabbitmqctl stop_app

# 重置本节点状态
rabbitmqctl reset

# 加入集群
rabbitmqctl join_cluster rabbit@mq-node1

# 启动 RabbitMQ 应用
rabbitmqctl start_app

# 验证集群状态
rabbitmqctl cluster_status

# 通过 CLI 创建
rabbitmqadmin declare queue \
  name=task.queue \
  durable=true \
  arguments='{"x-queue-type":"quorum"}'

channel.queue_declare(
    queue='task.queue',
    durable=True,
    arguments={
        'x-queue-type': 'quorum',
        # 初始副本数（默认等于集群节点数，最多 5）
        'x-quorum-initial-group-size': 3,
    }
)

# 创建 Vhost
rabbitmqctl add_vhost /production
rabbitmqctl add_vhost /staging

# 创建用户并授权
rabbitmqctl add_user app_user strong_password
rabbitmqctl set_permissions -p /production app_user ".*" ".*" ".*"
# 格式：set_permissions -p {vhost} {user} {configure正则} {write正则} {read正则}

# 查看 Vhost 权限
rabbitmqctl list_permissions -p /production

# 内存水位：当 RabbitMQ 使用内存超过系统总内存的 40% 时触发流控
vm_memory_high_watermark.relative = 0.4

# 也可以使用绝对值
# vm_memory_high_watermark.absolute = 4GB

# 磁盘空闲空间低于此值时触发流控（防止磁盘写满）
disk_free_limit.absolute = 5GB

# 或相对于内存的倍数
# disk_free_limit.relative = 2.0

rabbitmqctl status | grep -A5 "memory\|disk_free"

# 单个连接最大信道数（默认 2047）
channel_max = 200

# 最大连接数（默认无限制）
connection_max = 500

# 查看当前连接数
rabbitmqctl list_connections | wc -l

# 查看信道数
rabbitmqctl list_channels | wc -l

# 队列持久化（durable=True）
channel.queue_declare(queue='important.tasks', durable=True)

# 消息持久化（delivery_mode=2）
channel.basic_publish(
    exchange='',
    routing_key='important.tasks',
    body=json.dumps(message),
    properties=pika.BasicProperties(
        delivery_mode=2,          # 持久化
        content_type='application/json',
        message_id=str(uuid.uuid4()),
        timestamp=int(time.time()),
    )
)

# 队列级别：所有消息 TTL 为 1 小时
channel.queue_declare(
    queue='temp.tasks',
    durable=True,
    arguments={
        'x-message-ttl': 3600000,        # 毫秒
        'x-expires': 7200000,            # 队列空闲 2h 后自动删除
        'x-max-length': 100000,          # 最大消息数
        'x-max-length-bytes': 104857600, # 最大字节数（100MB）
        'x-overflow': 'reject-publish',  # 超出限制后拒绝新消息（而非丢弃旧消息）
    }
)

def process_message(ch, method, properties, body):
    try:
        data = json.loads(body)
        handle_task(data)
        # 处理成功，确认消息
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except RetryableError as e:
        logger.warning(f"可重试错误，消息重新入队: {e}")
        # 重新入队，但要注意无限循环问题
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
    except PermanentError as e:
        logger.error(f"永久性错误，消息进入死信队列: {e}")
        # 不重新入队，发往死信队列
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
    except Exception as e:
        logger.error(f"未知错误: {e}", exc_info=True)
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

# 设置 prefetch count
# 值为 1：严格逐条处理，吞吐量最低但最公平
# 值为 10~100：批量预取，吞吐量更高，但单个消费者可能积压更多未处理消息
channel.basic_qos(prefetch_count=10)

# 更合理的方式：按字节设置（需 RabbitMQ 3.x 的 global QoS 支持）
# 目前 basic_qos 的 prefetch_size 参数大多数客户端库不完整支持
# 建议通过 prefetch_count + 消息大小控制间接实现
channel.basic_qos(prefetch_count=20, global_qos=False)
# global_qos=True：限制整个信道
# global_qos=False（默认）：限制单个消费者

# 1. 先声明死信 Exchange 和队列
channel.exchange_declare(exchange='dlx', exchange_type='direct', durable=True)
channel.queue_declare(queue='task.queue.dead', durable=True)
channel.queue_bind(queue='task.queue.dead', exchange='dlx', routing_key='task.queue')

# 2. 创建业务队列，绑定 DLX
channel.queue_declare(
    queue='task.queue',
    durable=True,
    arguments={
        'x-queue-type': 'quorum',
        'x-dead-letter-exchange': 'dlx',         # 死信 Exchange
        'x-dead-letter-routing-key': 'task.queue', # 死信 Routing Key
        'x-delivery-limit': 3,                    # Quorum Queue: 最大投递次数
    }
)

# 死信消费者：记录、报警、人工处理或降级处理
def process_dead_letter(ch, method, properties, body):
    headers = properties.headers or {}
    death_info = headers.get('x-death', [{}])[0]

    logger.error(
        "消息进入死信队列",
        extra={
            'original_queue': death_info.get('queue'),
            'reason': death_info.get('reason'),
            'count': death_info.get('count'),
            'message_id': properties.message_id,
            'body_preview': body[:200].decode('utf-8', errors='replace'),
        }
    )

    # 根据情况：持久化到 DB、发告警、人工干预
    save_to_failed_messages_db(properties, body)
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

# 创建 delayed exchange
channel.exchange_declare(
    exchange='delayed.exchange',
    exchange_type='x-delayed-message',
    durable=True,
    arguments={'x-delayed-type': 'direct'}
)

# 发送延迟消息（延迟 30 秒）
channel.basic_publish(
    exchange='delayed.exchange',
    routing_key='task.delayed',
    body=json.dumps(payload),
    properties=pika.BasicProperties(
        headers={'x-delay': 30000},   # 毫秒
        delivery_mode=2,
    )
)

# 获取所有队列状态
curl -s -u guest:guest http://localhost:15672/api/queues/%2F | \
  jq '.[] | {name, messages, messages_ready, messages_unacknowledged, consumers}'

# 获取单个队列详情
curl -s -u admin:password \
  "http://localhost:15672/api/queues/%2Fproduction/task.queue" | jq .

# 获取节点状态
curl -s -u admin:password http://localhost:15672/api/nodes | \
  jq '.[] | {name, running, mem_used, disk_free, proc_used}'

# 获取连接列表
curl -s -u admin:password http://localhost:15672/api/connections | \
  jq '.[] | {name, state, channels, send_pend}'

rabbitmq-plugins enable rabbitmq_prometheus

# prometheus.yml 抓取配置
scrape_configs:
  - job_name: 'rabbitmq'
    static_configs:
      - targets:
        - 'mq-node1:15692'
        - 'mq-node2:15692'
        - 'mq-node3:15692'
    metric_relabel_configs:
      - source_labels: [queue]
        target_label: queue_name

# docker-compose.yml
services:
  rabbitmq_exporter:
    image: kbudde/rabbitmq-exporter:latest
    environment:
      RABBIT_URL: "http://mq-node1:15672"
      RABBIT_USER: monitoring
      RABBIT_PASSWORD: password
      RABBIT_CAPABILITIES: "bert,no_sort"
      PUBLISH_PORT: "9419"
    ports:
      - "9419:9419"

# prometheus-rules.yaml
groups:
  - name: rabbitmq.rules
    rules:
      # 队列消息积压
      - alert: RabbitmqQueueMessagesHigh
        expr: |
          rabbitmq_queue_messages{queue!~".*\\.dead"} > 50000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "队列 {{ $labels.queue }} 消息积压"
          description: "队列 {{ $labels.queue }} 积压 {{ $value }} 条消息"

      # 消费者数量为 0
      - alert: RabbitmqQueueNoConsumers
        expr: |
          rabbitmq_queue_consumers{queue!~".*\\.dead"} == 0
            and
          rabbitmq_queue_messages > 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "队列 {{ $labels.queue }} 无消费者"

      # 节点内存超水位
      - alert: RabbitmqMemoryHighWatermark
        expr: |
          rabbitmq_process_resident_memory_bytes
          /
          rabbitmq_vm_memory_high_watermark_bytes > 0.9
        for: 3m
        labels:
          severity: warning
        annotations:
          summary: "RabbitMQ 节点 {{ $labels.instance }} 内存接近水位"

      # 节点离线
      - alert: RabbitmqNodeDown
        expr: up{job="rabbitmq"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "RabbitMQ 节点 {{ $labels.instance }} 不可达"

      # 死信队列有消息积压
      - alert: RabbitmqDeadLetterQueueNotEmpty
        expr: rabbitmq_queue_messages{queue=~".*\\.dead"} > 0
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "死信队列 {{ $labels.queue }} 有 {{ $value }} 条消息需要处理"

# 查看各队列积压情况
rabbitmqctl list_queues name messages consumers message_bytes \
  --vhost /production \
  --formatter table

# 找出积压最严重的队列
rabbitmqctl list_queues name messages consumers --formatter table | \
  sort -k2 -rn | head -20

# 方法1：临时增加 prefetch（在消费者配置中调大）
# 方法2：手动 purge 非关键队列（慎用！消息会丢失）
rabbitmqctl purge_queue non_critical_queue --vhost /production

# 方法3：将消息转移到另一个队列（使用 shovel 插件）
rabbitmq-plugins enable rabbitmq_shovel rabbitmq_shovel_management

# 查看 Quorum Queue 副本状态
rabbitmq-diagnostics check_if_any_deprecated_features_are_used

# 查看详细副本分布
rabbitmqctl list_queues name type members online slave_pids \
  --vhost /production \
  --formatter table

# 手动触发 Quorum Queue 成员重选（一般不需要）
rabbitmqctl force_boot  # 仅在所有节点都下线时使用，否则可能丢数据！

# 查看未完全同步的镜像队列
rabbitmqctl list_queues name slave_pids synchronised_slave_pids \
  --vhost /production | \
  awk '{if($2!=$3) print $0}'

# 手动触发同步
rabbitmqctl sync_queue -p /production queue_name

rabbitmqctl cluster_status | grep -A5 "Network Partitions"

# 1. 确认网络已恢复，所有节点互通
ping mq-node1
ping mq-node2
ping mq-node3

# 2. 检查分区状态
rabbitmqctl cluster_status

# 3. 如果还有分区记录，需要滚动重启节点来清除分区状态
# 先重启少数派节点（它们的数据更旧）
systemctl restart rabbitmq-server  # 在少数派节点执行

# 4. 验证集群状态
rabbitmqctl cluster_status

# 在其中一个节点执行，强制重新合并（会触发数据选择）
rabbitmqctl force_reset  # 极端情况才用，会清空该节点数据！

# 更安全的方式：移除再重新加入集群
rabbitmqctl stop_app
rabbitmqctl reset
rabbitmqctl join_cluster rabbit@mq-node1
rabbitmqctl start_app

# 查看被阻塞的连接
rabbitmqctl list_connections name state blocked_by --formatter table

# 查看当前内存使用
rabbitmqctl status | grep memory

# 查看磁盘使用
rabbitmqctl status | grep disk_free

# 临时提高内存水位（治标不治本）
rabbitmqctl set_vm_memory_high_watermark 0.5

# 快速消费或清理队列，释放内存
rabbitmqctl purge_queue large_queue --vhost /production

# 查看集群状态
rabbitmqctl cluster_status

# 查看节点信息
rabbitmqctl node_health_check
rabbitmq-diagnostics check_port_connectivity
rabbitmq-diagnostics check_protocol_listener

# 优雅关闭节点（等待消费者完成当前消息）
rabbitmqctl shutdown

# 强制关闭（紧急情况）
rabbitmqctl stop_app

# 列出用户
rabbitmqctl list_users

# 添加用户
rabbitmqctl add_user new_user strong_password

# 设置角色（administrator/monitoring/policymaker/management）
rabbitmqctl set_user_tags new_user monitoring

# 设置 Vhost 权限
rabbitmqctl set_permissions -p /production new_user ".*" ".*" ".*"

# 删除用户
rabbitmqctl delete_user old_user

# 修改密码
rabbitmqctl change_password myuser new_password

# 列出队列（含消息数和消费者数）
rabbitmqctl list_queues name messages consumers memory --vhost /production

# 删除队列
rabbitmqctl delete_queue my_queue --vhost /production

# 清空队列（消息丢失，慎用）
rabbitmqctl purge_queue my_queue --vhost /production

# 查看队列详情
rabbitmqadmin get queue=my_queue count=1 --vhost=/production

# 列出 Exchange
rabbitmqctl list_exchanges name type durable --vhost /production

# 列出 Binding
rabbitmqctl list_bindings \
  source_name destination_name routing_key \
  --vhost /production

# 通过 rabbitmqadmin 声明 Exchange
rabbitmqadmin declare exchange \
  name=my.exchange \
  type=topic \
  durable=true \
  --vhost=/production

# 给所有队列添加死信队列策略
rabbitmqctl set_policy DLX ".*" \
  '{"dead-letter-exchange":"dlx"}' \
  --apply-to queues \
  --vhost /production \
  --priority 10

# 给特定队列设置 TTL
rabbitmqctl set_policy TTL "temp\\..*" \
  '{"message-ttl":3600000}' \
  --apply-to queues \
  --vhost /production

# 查看所有策略
rabbitmqctl list_policies --vhost /production

# 删除策略
rabbitmqctl clear_policy DLX --vhost /production

# 实时查看 RabbitMQ 日志
tail -f /var/log/rabbitmq/rabbit@mq-node1.log

# 查看连接详情
rabbitmqctl list_connections \
  name peer_host peer_port state channels \
  send_pend recv_cnt send_cnt

# 查看信道信息
rabbitmqctl list_channels \
  connection name consumer_count messages_unacknowledged

# 查看消费者信息
rabbitmqctl list_consumers \
  --vhost /production \
  queue_name channel_pid consumer_tag prefetch_count

# 查看插件状态
rabbitmq-plugins list
rabbitmq-plugins list --enabled

# 启用/禁用插件
rabbitmq-plugins enable rabbitmq_shovel
rabbitmq-plugins disable rabbitmq_mqtt

# 安装
curl -s http://localhost:15672/cli/rabbitmqadmin > /usr/local/bin/rabbitmqadmin
chmod +x /usr/local/bin/rabbitmqadmin

# 配置别名（含认证）
alias rmqadm='rabbitmqadmin -u admin -p password -V /production'

# 发布测试消息
rabbitmqadmin publish \
  exchange=amq.default \
  routing_key=test.queue \
  payload='{"test": true}' \
  -u admin -p password

# 获取（消费）一条消息
rabbitmqadmin get queue=test.queue count=1 -u admin -p password

# 导出所有配置（备份）
rabbitmqadmin export /backup/rabbitmq-config-$(date +%Y%m%d).json \
  -u admin -p password

# 导入配置（恢复）
rabbitmqadmin import /backup/rabbitmq-config-20240422.json \
  -u admin -p password