import pika

# 基础连接
connection = pika.BlockingConnection(
    pika.ConnectionParameters(
        host='localhost',
        port=5672,
        virtual_host='/',
        credentials=pika.PlainCredentials('admin', 'admin123'),
        heartbeat=60,
        blocked_connection_timeout=300,
        connection_attempts=3,
        retry_delay=5
    )
)

import pika
from contextlib import contextmanager

class RabbitMQPool:
    def __init__(self, host, port, vhost, username, password, pool_size=5):
        self.params = pika.ConnectionParameters(
            host=host, port=port, virtual_host=vhost,
            credentials=pika.PlainCredentials(username, password),
            heartbeat=60
        )
        self.pool_size = pool_size
        self._connections = []

    def _create_connection(self):
        return pika.BlockingConnection(self.params)

    @contextmanager
    def get_connection(self):
        if self._connections:
            conn = self._connections.pop()
            if conn.is_open:
                try:
                    yield conn
                    self._connections.append(conn)
                    return
                except Exception:
                    conn.close()
        conn = self._create_connection()
        try:
            yield conn
            if len(self._connections) < self.pool_size:
                self._connections.append(conn)
            else:
                conn.close()
        except Exception:
            conn.close()
            raise

# 使用
pool = RabbitMQPool('localhost', 5672, '/', 'admin', 'admin123')
with pool.get_connection() as conn:
    channel = conn.channel()
    channel.queue_declare(queue='test')

# 设置预取数量（QoS）
channel.basic_qos(prefetch_count=10)

# 开启 Publisher Confirm（发布确认）
channel.confirm_delivery()

# ❌ 错误：多线程共享 Channel
# ✅ 正确：每线程一个 Channel
def worker(connection):
    channel = connection.channel()
    # 使用 channel 进行操作
    channel.close()

# Direct Exchange
channel.exchange_declare(
    exchange='order_direct',
    exchange_type='direct',
    durable=True,      # 持久化
    auto_delete=False   # 不自动删除
)

# Fanout Exchange
channel.exchange_declare(
    exchange='notification_fanout',
    exchange_type='fanout',
    durable=True
)

# Topic Exchange
channel.exchange_declare(
    exchange='log_topic',
    exchange_type='topic',
    durable=True
)

result = channel.queue_declare(
    queue='order_queue',
    durable=True,          # 持久化
    exclusive=False,       # 非排他
    auto_delete=False,     # 不自动删除
    arguments={
        'x-message-ttl': 60000,           # 消息 TTL（毫秒）
        'x-max-length': 10000,            # 最大消息数
        'x-max-length-bytes': 104857600,  # 最大字节数（100MB）
        'x-overflow': 'reject-publish',   # 溢出策略
        'x-dead-letter-exchange': 'dlx',  # 死信交换机
        'x-dead-letter-routing-key': 'dead'  # 死信路由键
    }
)

# 获取服务端生成的队列名（匿名队列）
queue_name = result.method.queue

# 将队列绑定到 Direct Exchange，指定路由键
channel.queue_bind(
    exchange='order_direct',
    queue='payment_queue',
    routing_key='order.payment'
)

# 将队列绑定到 Fanout Exchange（无需路由键）
channel.queue_bind(
    exchange='notification_fanout',
    queue='email_queue'
)

# 将队列绑定到 Topic Exchange
channel.queue_bind(
    exchange='log_topic',
    queue='error_queue',
    routing_key='*.error.#'
)

# Exchange 之间的绑定（Exchange-to-Exchange）
channel.exchange_bind(
    destination='log_fanout',
    source='log_topic',
    routing_key='#.error.#'
)

# 创建
rabbitmqctl add_vhost production

# 删除
rabbitmqctl delete_vhost production

# 列出所有
rabbitmqctl list_vhosts

# 设置权限
rabbitmqctl set_permissions -p production user1 ".*" ".*" ".*"
rabbitmqctl set_permissions -p production user2 "^order-.*" "^order-.*" "^order-.*"

# 查看权限
rabbitmqctl list_permissions -p production

import com.rabbitmq.client.*;

public class Producer {
    private static final String EXCHANGE_NAME = "order_direct";
    private static final String QUEUE_NAME = "payment_queue";
    private static final String ROUTING_KEY = "order.payment";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin123");
        factory.setVirtualHost("/");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true);

            // 声明队列
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

            // 绑定
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

            // 发布消息
            String message = "{\"order_id\":\"20260510001\",\"amount\":99.9}";
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                    .contentType("application/json")
                    .deliveryMode(2)  // 持久化
                    .build();

            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, props, message.getBytes());
            System.out.println("[x] 消息已发送: " + message);
        }
    }
}

import com.rabbitmq.client.*;

public class Consumer {
    private static final String QUEUE_NAME = "payment_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("admin");
        factory.setPassword("admin123");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 设置 QoS
        channel.basicQos(10);

        // 消费消息
        DeliverCallback deliverCallback = (tag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("[x] 收到消息: " + message);
            // 处理消息...
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };

        CancelCallback cancelCallback = tag -> {
            System.out.println("[!] 消费被取消: " + tag);
        };

        channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);
        System.out.println("[*] 等待消息中...");
    }
}

package main

import (
    "log"
    amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
    // 连接
    conn, err := amqp.Dial("amqp://admin:admin123@localhost:5672/")
    if err != nil {
        log.Fatalf("连接失败: %v", err)
    }
    defer conn.Close()

    // 创建 Channel
    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("创建 Channel 失败: %v", err)
    }
    defer ch.Close()

    // 声明 Exchange
    err = ch.ExchangeDeclare(
        "order_direct", "direct",
        true, false, false, false, nil,
    )
    if err != nil {
        log.Fatalf("声明 Exchange 失败: %v", err)
    }

    // 声明队列
    q, err := ch.QueueDeclare(
        "payment_queue", true, false, false, nil,
    )
    if err != nil {
        log.Fatalf("声明队列失败: %v", err)
    }

    // 绑定
    err = ch.QueueBind(q.Name, "order.payment", "order_direct", false, nil)
    if err != nil {
        log.Fatalf("绑定失败: %v", err)
    }

    // 发布消息
    err = ch.Publish("order_direct", "order.payment", false, false,
        amqp.Publishing{
            ContentType:  "application/json",
            DeliveryMode: amqp.Persistent,
            Body:         []byte(`{"order_id":"20260510001","amount":99.9}`),
        })
    if err != nil {
        log.Fatalf("发布失败: %v", err)
    }

    log.Println("[x] 消息已发送")
}