RabbitMq磁盘空间不足对消息处理的影响,解析消息中间件因存储耗尽导致的服务崩溃问题

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建持久化队列（关键参数）
channel.queue_declare(queue='order_queue', durable=True)

# 发送持久化消息（关键参数）
for i in range(100000):
    channel.basic_publish(
        exchange='',
        routing_key='order_queue',
        body=f'订单_{i}',
        properties=pika.BasicProperties(
            delivery_mode=2,  # 持久化标志
        ))

# 消费者处理异常示例
def callback(ch, method, properties, body):
    try:
        # 模拟处理耗时操作
        time.sleep(0.1) 
        print(f"处理完成: {body.decode()}")
    except Exception as e:
        # 未捕获异常会导致消息重新入队
        print("处理失败，消息重新排队")
        ch.basic_nack(delivery_tag=method.delivery_tag)

channel.basic_consume(
    queue='order_queue',
    on_message_callback=callback,
    auto_ack=False)

#!/bin/bash
# 磁盘空间监控脚本
THRESHOLD=90  # 磁盘使用百分比阈值
CURRENT=$(df /var/lib/rabbitmq | awk 'NR==2 {print $5}' | sed 's/%//')

if [ $CURRENT -ge $THRESHOLD ]; then
    # 自动触发消息清理
    rabbitmqctl purge_queue order_queue
    # 发送报警通知
    curl -X POST https://api.alert.com -d 'disk_full_alert'
fi

# 自动消息清理工具
import pika

class QueueCleaner:
    def __init__(self):
        self.connection = pika.BlockingConnection(...)
        
    def smart_purge(self, queue_name):
        # 获取队列深度
        queue = self.channel.queue_declare(queue_name, passive=True)
        msg_count = queue.method.message_count
        
        # 保留最后100条消息
        if msg_count > 100:
            self.channel.queue_purge(queue_name)
            print(f"已清理 {msg_count} 条消息")
        else:
            print("无需清理")

# 混合存储优化示例
channel.basic_publish(
    exchange='',
    routing_key='mixed_queue',
    body=large_data,
    properties=pika.BasicProperties(
        delivery_mode=1,  # 非持久化
        headers={
            'storage_type': 'external_db',  # 元数据标记
            'ref_id': '123456'  # 外键关联
        }
    ))

# 自定义健康检查端点
curl -s http://localhost:15672/api/healthchecks/node

{
    "status": "ok",
    "disk_free": 1073741824,
    "mem_used": 536870912
}

# Docker集群部署示例
version: '3'
services:
  rabbit1:
    image: rabbitmq:3.9-management
    environment:
      - RABBITMQ_ERLANG_COOKIE=secret
      - RABBITMQ_DEFAULT_USER=admin
    volumes:
      - ./data1:/var/lib/rabbitmq

  rabbit2:
    image: rabbitmq:3.9-management
    links:
      - rabbit1
    environment:
      - RABBITMQ_ERLANG_COOKIE=secret
      - RABBITMQ_DEFAULT_USER=admin
    command: bash -c "sleep 10 && rabbitmqctl stop_app && \
             rabbitmqctl join_cluster rabbit@rabbit1 && \
             rabbitmqctl start_app"