channel.queue_declare(
    queue='classic_queue',
    durable=True,        # 持久化队列
    exclusive=False,     # 非排他
    auto_delete=False,   # 不自动删除
    arguments={
        'x-queue-type': 'classic'  # 显式指定类型
    }
)

import pika

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

# 1. 持久化交换机
channel.exchange_declare(
    exchange='persistent_exchange',
    exchange_type='direct',
    durable=True  # 持久化
)

# 2. 持久化队列
channel.queue_declare(
    queue='persistent_queue',
    durable=True  # 持久化
)

channel.queue_bind(
    exchange='persistent_exchange',
    queue='persistent_queue',
    routing_key='task'
)

# 3. 持久化消息
channel.basic_publish(
    exchange='persistent_exchange',
    routing_key='task',
    body='Important data',
    properties=pika.BasicProperties(
        delivery_mode=2  # 持久化消息
    )
)

# 客户端
result = channel.queue_declare(queue='', exclusive=True)
callback_queue = result.method.queue

# 发送 RPC 请求
channel.basic_publish(
    exchange='',
    routing_key='rpc_queue',
    properties=pika.BasicProperties(
        reply_to=callback_queue,        # 告知服务端回复到此队列
        correlation_id=str(uuid.uuid4()) # 关联 ID
    ),
    body=json.dumps({'method': 'calculate', 'params': [1, 2, 3]})
)

# 等待回复
channel.basic_consume(queue=callback_queue, on_message_callback=on_response)

channel.queue_declare(
    queue='temp_notification',
    auto_delete=True
)

# 步骤 1: 创建死信交换机和死信队列
channel.exchange_declare(exchange='dlx', exchange_type='direct', durable=True)
channel.queue_declare(queue='dead_letters', durable=True)
channel.queue_bind(exchange='dlx', queue='dead_letters', routing_key='dead')

# 步骤 2: 业务队列绑定死信交换机
channel.queue_declare(
    queue='order_queue',
    durable=True,
    arguments={
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'dead',
        'x-message-ttl': 60000,           # 消息 TTL 60 秒
        'x-max-length': 5000,             # 最大 5000 条
        'x-overflow': 'reject-publish'    # 超出时拒绝发布
    }
)

import pika
import json

MAX_RETRIES = 3

def process_message(ch, method, properties, body):
    message = json.loads(body)
    retry_count = properties.headers.get('x-retry-count', 0) if properties.headers else 0

    try:
        # 业务处理
        handle_business(message)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        if retry_count < MAX_RETRIES:
            # 重新发布，增加重试计数
            headers = dict(properties.headers or {})
            headers['x-retry-count'] = retry_count + 1
            ch.basic_publish(
                exchange='retry_exchange',
                routing_key='retry',
                body=body,
                properties=pika.BasicProperties(
                    delivery_mode=2,
                    headers=headers
                )
            )
            ch.basic_ack(delivery_tag=method.delivery_tag)
        else:
            # 超过重试次数，拒绝并进入死信队列
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

channel.queue_declare(
    queue='priority_queue',
    durable=True,
    arguments={
        'x-max-priority': 10  # 最大优先级（0-255，推荐 1-10）
    }
)

# 高优先级消息
channel.basic_publish(
    exchange='task_exchange',
    routing_key='task',
    body=json.dumps({'task': 'urgent_refund', 'priority': 'high'}),
    properties=pika.BasicProperties(
        delivery_mode=2,
        priority=10  # 高优先级
    )
)

# 低优先级消息
channel.basic_publish(
    exchange='task_exchange',
    routing_key='task',
    body=json.dumps({'task': 'generate_report', 'priority': 'low'}),
    properties=pika.BasicProperties(
        delivery_mode=2,
        priority=1  # 低优先级
    )
)

# 方式 1: 队列声明时设置
channel.queue_declare(
    queue='lazy_queue',
    durable=True,
    arguments={'x-queue-mode': 'lazy'}
)

# 方式 2: 通过策略设置（推荐）
# rabbitmqctl set_policy lazy-queues "^lazy\." '{"queue-mode":"lazy"}' --apply-to queues

channel.queue_declare(
    queue='orders',
    durable=True,
    arguments={
        'x-queue-type': 'quorum',
        'x-quorum-initial-group-size': 3,   # 初始组大小（可选）
        'x-delivery-limit': 5,              # 消息最大投递次数
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'dead',
        'x-message-ttl': 300000,
        'x-max-length': 100000,
        'x-max-length-bytes': 1073741824    # 1GB
    }
)

# 查看仲裁队列状态
rabbitmq-queues quorum_status "orders"

# 查看所有仲裁队列成员
rabbitmq-queues members "orders"

# 添加成员（扩展集群时）
rabbitmq-queues add_member orders rabbit@node4

# 移除成员
rabbitmq-queues remove_member orders rabbit@node3

# 检查集群健康
rabbitmq-diagnostics check_if_node_is_quorum_critical

channel.queue_declare(
    queue='ttl_queue',
    arguments={'x-message-ttl': 60000}  # 所有消息 60 秒后过期
)

channel.basic_publish(
    exchange='',
    routing_key='ttl_queue',
    body='short-lived message',
    properties=pika.BasicProperties(
        expiration='10000'  # 这条消息 10 秒后过期（字符串，毫秒）
    )
)

channel.queue_declare(
    queue='bounded_queue',
    arguments={
        'x-max-length': 10000,
        'x-max-length-bytes': 104857600,  # 100MB
        'x-overflow': 'reject-publish'
    }
)