import pika

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

# 声明队列（自动绑定到默认交换机，路由键 = 队列名）
channel.queue_declare(queue='task_queue', durable=True)

# 使用默认交换机（exchange=''），路由键为队列名
channel.basic_publish(
    exchange='',                    # 默认交换机
    routing_key='task_queue',       # 路由键 = 队列名
    body='Hello World!',
    properties=pika.BasicProperties(delivery_mode=2)
)

connection.close()

import pika
import json

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

# 声明 Direct 交换机
channel.exchange_declare(exchange='order_direct', exchange_type='direct', durable=True)

# 声明多个队列
for queue_name in ['payment_queue', 'shipping_queue', 'email_queue']:
    channel.queue_declare(queue=queue_name, durable=True)

# 绑定队列到交换机
channel.queue_bind(exchange='order_direct', queue='payment_queue', routing_key='payment')
channel.queue_bind(exchange='order_direct', queue='shipping_queue', routing_key='shipping')
channel.queue_bind(exchange='order_direct', queue='email_queue', routing_key='email')

# 发布消息 - 路由到 payment_queue
channel.basic_publish(
    exchange='order_direct',
    routing_key='payment',
    body=json.dumps({'order_id': '001', 'amount': 299.9}),
    properties=pika.BasicProperties(delivery_mode=2)
)

print("[x] 支付消息已发送")
connection.close()

import pika
import json

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

# 声明 Fanout 交换机
channel.exchange_declare(exchange='user_events', exchange_type='fanout', durable=True)

# 声明各服务的队列
channels = ['sms_queue', 'email_queue', 'push_queue']
for q in channels:
    channel.queue_declare(queue=q, durable=True)
    channel.queue_bind(exchange='user_events', queue=q)

# 发布用户注册事件
event = {
    'event': 'user_registered',
    'user_id': 'u_1001',
    'username': 'zhangsan',
    'email': '[email protected]',
    'phone': '13800138000'
}

channel.basic_publish(
    exchange='user_events',
    routing_key='',        # Fanout 忽略 routing_key
    body=json.dumps(event),
    properties=pika.BasicProperties(delivery_mode=2)
)

print("[x] 用户注册事件已广播")
connection.close()

# 消费者使用临时队列接收广播
result = channel.queue_declare(queue='', exclusive=True)  # 匿名排他队列
queue_name = result.method.queue

channel.queue_bind(exchange='user_events', queue=queue_name)

def callback(ch, method, properties, body):
    print(f"[x] 收到事件: {body.decode()}")

channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
channel.start_consuming()

import pika
import json
from datetime import datetime

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

# 声明 Topic 交换机
channel.exchange_declare(exchange='app_logs', exchange_type='topic', durable=True)

# 声明队列
queues_config = {
    'error_queue':     '#.ERROR.#',       # 所有错误日志
    'warning_queue':   '#.WARNING.#',     # 所有警告日志
    'payment_queue':   'payment.#',       # 支付相关日志
    'critical_queue':  '*.CRITICAL.#',    # 临界错误
    'all_queue':       '#',               # 所有日志
}

for queue_name, binding_key in queues_config.items():
    channel.queue_declare(queue=queue_name, durable=True)
    channel.queue_bind(exchange='app_logs', queue=queue_name, routing_key=binding_key)

# 发送日志
logs = [
    ('payment.INFO', '用户支付成功'),
    ('payment.ERROR', '支付网关超时'),
    ('order.WARNING', '库存不足'),
    ('system.CRITICAL', '数据库连接失败'),
    ('user.INFO', '用户登录'),
]

for routing_key, message in logs:
    log_entry = {
        'level': routing_key.split('.')[1],
        'service': routing_key.split('.')[0],
        'message': message,
        'timestamp': datetime.now().isoformat()
    }
    channel.basic_publish(
        exchange='app_logs',
        routing_key=routing_key,
        body=json.dumps(log_entry),
        properties=pika.BasicProperties(delivery_mode=2)
    )
    print(f"[x] {routing_key}: {message}")

connection.close()

import pika

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

# 声明 Headers 交换机
channel.exchange_declare(exchange='doc_headers', exchange_type='headers', durable=True)

# 声明队列
channel.queue_declare(queue='pdf_queue', durable=True)
channel.queue_declare(queue='urgent_queue', durable=True)

# 绑定 - 匹配 format=pdf 的文档
channel.queue_bind(
    exchange='doc_headers', queue='pdf_queue',
    arguments={'x-match': 'all', 'format': 'pdf'}
)

# 绑定 - 匹配 priority=high 或 format=pdf
channel.queue_bind(
    exchange='doc_headers', queue='urgent_queue',
    arguments={'x-match': 'any', 'priority': 'high', 'format': 'pdf'}
)

# 发布消息（通过 headers 路由）
channel.basic_publish(
    exchange='doc_headers',
    routing_key='',  # Headers 交换机忽略 routing_key
    body=b'PDF document content',
    properties=pika.BasicProperties(
        delivery_mode=2,
        headers={'format': 'pdf', 'priority': 'high', 'language': 'zh'}
    )
)

print("[x] 文档消息已发送")
connection.close()

import pika

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

# 1. 声明死信交换机和死信队列
channel.exchange_declare(exchange='dlx_exchange', exchange_type='direct', durable=True)
channel.queue_declare(queue='dead_letter_queue', durable=True)
channel.queue_bind(exchange='dlx_exchange', queue='dead_letter_queue', routing_key='dead')

# 2. 声明业务队列（配置死信交换机）
channel.queue_declare(
    queue='business_queue',
    durable=True,
    arguments={
        'x-dead-letter-exchange': 'dlx_exchange',
        'x-dead-letter-routing-key': 'dead',
        'x-message-ttl': 30000,         # 消息 30 秒未消费成为死信
        'x-max-length': 1000            # 超出 1000 条成为死信
    }
)

channel.queue_bind(exchange='business_exchange', queue='business_queue', routing_key='task')

# 死信队列消费者 - 记录/告警/重试
def dead_letter_handler(ch, method, properties, body):
    print(f"[DLX] 收到死信: {body.decode()}")
    print(f"  原交换机: {properties.headers.get('x-first-death-exchange', 'N/A')}")
    print(f"  原队列: {properties.headers.get('x-first-death-queue', 'N/A')}")
    print(f"  死因: {properties.headers.get('x-first-death-reason', 'N/A')}")
    
    # 记录到数据库 / 发送告警 / 尝试重试
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='dead_letter_queue', on_message_callback=dead_letter_handler)

# 下载插件（版本需匹配 RabbitMQ 版本）
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.13.0/rabbitmq_delayed_message_exchange-3.13.0.ez
cp rabbitmq_delayed_message_exchange-3.13.0.ez /opt/rabbitmq/plugins/
rabbitmq-plugins enable rabbitmq_delayed_message_exchange

import pika
import json
import time

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

# 声明延迟交换机（type=x-delayed-message）
args = {'x-delayed-type': 'direct'}  # 内部路由类型
channel.exchange_declare(
    exchange='delayed_exchange',
    exchange_type='x-delayed-message',
    durable=True,
    arguments=args
)

channel.queue_declare(queue='delayed_queue', durable=True)
channel.queue_bind(exchange='delayed_exchange', queue='delayed_queue', routing_key='delay')

# 发送延迟消息（延迟 5 秒）
channel.basic_publish(
    exchange='delayed_exchange',
    routing_key='delay',
    body=json.dumps({
        'task': 'send_reminder',
        'user_id': 'u_1001',
        'message': '您的订单即将超时，请尽快支付'
    }),
    properties=pika.BasicProperties(
        delivery_mode=2,
        headers={'x-delay': 5000}  # 延迟 5000 毫秒
    )
)

print(f"[x] 延迟消息已发送，将在 5 秒后投递")
connection.close()

# 声明备用交换机
channel.exchange_declare(exchange='fallback_exchange', exchange_type='fanout', durable=True)
channel.queue_declare(queue='unroutable_queue', durable=True)
channel.queue_bind(exchange='fallback_exchange', queue='unroutable_queue')

# 声明主交换机，指定备用交换机
channel.exchange_declare(
    exchange='main_exchange',
    exchange_type='direct',
    durable=True,
    arguments={'alternate-exchange': 'fallback_exchange'}
)

channel.exchange_declare(
    exchange='internal_router',
    exchange_type='topic',
    durable=True,
    internal=True  # 标记为内部交换机
)