channel.exchangeDeclare(
    "order_exchange",   // exchange: 交換機名稱
    "direct",           // type: 交換機類型
    true,               // durable: 是否持久化
    false,              // autoDelete: 自動刪除
    null                // arguments: 擴展參數(shù)
);

channel.queueDeclare(
    "order_queue",      // queue: 隊列名稱
    true,               // durable: 持久化
    false,              // exclusive: 是否獨占
    false,              // autoDelete: 自動刪除
    new HashMap<String, Object>(){{
        put("x-message-ttl", 60000); // 消息60秒過期
    }}                  // arguments: 隊列特性
);

// 創(chuàng)建延遲隊列（通過TTL+死信交換機）
Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 300000);       // 5分鐘過期
args.put("x-dead-letter-exchange", "dlx"); // 死信交換機
args.put("x-dead-letter-routing-key", "dead.orders"); // 死信路由鍵
// 創(chuàng)建優(yōu)先級隊列
args.put("x-max-priority", 10);          // 支持10級優(yōu)先級
// 限制隊列長度
args.put("x-max-length", 1000);          // 最多1000條消息

channel.queueBind(
    "email_queue",      // queue: 要綁定的隊列
    "notify_exchange",  // exchange: 交換機名稱
    "order.paid",       // routingKey: 路由鍵
    null                // arguments: 綁定參數(shù)
);

// 構建消息屬性
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .contentType("application/json")   // 內容類型
    .priority(5)                      // 優(yōu)先級（0-9）
    .deliveryMode(2)                  // 2=持久化（存硬盤）
    .expiration("60000")              // 消息60秒過期
    .build();
// 發(fā)送消息
channel.basicPublish(
    "order_exchange",   // exchange: 目標交換機
    "order.create",     // routingKey: 路由鍵
    props,              // 消息屬性（metadata）
    orderJson.getBytes() // 消息體內容（二進制）
);

// 錯誤！直接發(fā)到隊列（繞過交換機）-> 失去路由靈活性
channel.basicPublish("", "order_queue", null, message); 
// 正確！通過交換機路由
channel.basicPublish("order_exchange", "order.create", null, message);

// 創(chuàng)建消費者回調
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    try {
        // 處理消息（模擬業(yè)務邏輯）
        processOrder(message);
        // 手動確認 - 業(yè)務成功完成
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        // 處理失?。壕芙^消息（true=重新入隊重試）
        channel.basicReject(delivery.getEnvelope().getDeliveryTag(), true);
    }
};
// 開始消費
channel.basicConsume(
    "order_queue",   // queue: 監(jiān)聽的隊列
    false,           // autoAck: 關閉自動確認！！
    deliverCallback,  // 消息處理回調
    consumerTag -> {} // 取消訂閱時的回調（可忽略）
);

// 成功處理：確認單條消息
basicAck(long deliveryTag, boolean multiple); 
// multiple: true=批量確認之前所有消息（慎用！）
// 拒絕單條消息
basicReject(long deliveryTag, boolean requeue);
// requeue: true=重新入隊（可重試）, false=丟棄/進入死信
// 批量拒絕
basicNack(long deliveryTag, boolean multiple, boolean requeue);

// =========== 生產者端 ===========
// 1. 聲明持久化直連交換機
channel.exchangeDeclare("order_exchange", "direct", true);
// 2. 聲明持久化隊列（附加死信設置）
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "order_dlx");
channel.queueDeclare("order_queue", true, false, false, args);
// 3. 綁定隊列到交換機
channel.queueBind("order_queue", "order_exchange", "order.create");
// 4. 發(fā)送訂單消息（持久化）
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
        .deliveryMode(2)
        .build();
channel.basicPublish("order_exchange", "order.create", props, orderData);
// =========== 消費者端 ===========
// 1. 設置QoS：每次最多取5條（防止堆積）
channel.basicQos(5); 
// 2. 定義消息處理器
DeliverCallback callback = (tag, delivery) -> {
    try {
        handleOrder(delivery.getBody());
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (BusyException e) {
        // 系統(tǒng)繁忙：延遲重試
        Thread.sleep(5000);
        channel.basicReject(deliveryTag, true); 
    } catch (FatalException e) {
        // 致命錯誤：不再重試（進入死信）
        channel.basicReject(deliveryTag, false); 
    }
};
// 3. 開始消費（手動確認）
channel.basicConsume("order_queue", false, callback, tag -> {});

exchangeDeclare(..., true, ...)     // 交換機持久化
queueDeclare(..., true, ...)        // 隊列持久化
basicPublish(..., props.setDeliveryMode(2), ...) // 消息持久化

basicConsume(..., false, ...)       // 禁用autoAck
basicQos(prefetchCount)              // 設置預取數(shù)量

// 網(wǎng)絡重連
factory.setAutomaticRecoveryEnabled(true); 
// 業(yè)務重試
basicReject(..., true); // 重新入隊
// 死信隊列兜底
basicReject(..., false); 

// 開啟批量確認（減少IO）
channel.basicAck(lastDeliveryTag, true); 
// 使用Publisher Confirms（生產端）
channel.confirmSelect();