import pika
 
conn = pika.BlockingConnection()
ch   = conn.channel()
 
# 1. 死信交换机（复用）
ch.exchange_declare('dlx.exchange', exchange_type='direct', durable=True)
 
# 2. 死信队列（每个业务独享）
dlq_name = 'dlq.order.pay-ttl'
ch.queue_declare(dlq_name, durable=True)
ch.queue_bind(dlq_name, 'dlx.exchange', routing_key='rk.order.pay-ttl')
 
# 3. 业务队列（带 DLX 配置）
biz_q = 'q.order.pay-ttl'
args = {
    'x-message-ttl': 30 * 60 * 1000,          # 30 分钟 TTL
    'x-dead-letter-exchange': 'dlx.exchange',
    'x-dead-letter-routing-key': 'rk.order.pay-ttl'
}
ch.queue_declare(biz_q, durable=True, arguments=args)
 
# 4. 死信消费者（监听死信队列）
def on_dlq(ch, method, props, body):
    order_id = body.decode()
    # 查询微信 → 关闭或补偿
    ch.basic_ack(method.delivery_tag)
 
ch.basic_consume(dlq_name, on_dlq, auto_ack=False)
ch.start_consuming()

private static void executeTransaction(Channel channel) throws IOException {
        boolean transactionSuccess = false;
        try {
            // 开启事务
            channel.txSelect(); 
 
            // 执行一系列消息操作，例如：channel.basicPublish(exchange, routingKey, message);
            // 提交事务
            channel.txCommit(); 
            transactionSuccess = true;
        } catch (ShutdownSignalException | IOException e) {
            // 回滚事务
            if (!transactionSuccess) {
                channel.txRollback(); 
            }
            throw e;
        }
    }

public static void sendPersistentMessage(String host, String queueName, String message) {
        try (Connection connection = new ConnectionFactory().setHost(host).newConnection();
             Channel channel = connection.createChannel()) {
            // 启用发布者确认
            channel.confirmSelect(); 
 
           // 将消息设置为持久化
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2) 
                    .build();
                    
            // 添加确认监听器
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("消息已确认: " + deliveryTag);
                    // 消息正确到达Broker时的处理逻辑
                }
 
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("消息未确认: " + deliveryTag);
                    // 因为内部错误导致消息丢失时的处理逻辑
                }
            });
 
            channel.basicPublish("", queueName, properties, message.getBytes());
 
            // 等待消息确认，或者超时
            boolean allConfirmed = channel.waitForConfirms();
            
            if (allConfirmed) {
                //所有消息都已确认
            } else {
                //超时或其它
            }
           
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
}

// 设置 durable = true; 
channel.exchangeDeclare(exchangeName, "direct", durable);

// 设置 MessageProperties.PERSISTENT_TEXT_PLAIN
channel.basicPublish(exchangeName, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

//设置 boolean durable = true;
channel.queueDeclare(queueName, durable, exclusive, false, null);

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    try {
        //接收消息，业务处理
        //设置手动确认
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        //发生异常时，可以选择重新发送消息或进行错误处理
        // 例如，可以选择负确认（nack），让消息重回队列
        // channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
};
//设置autoAck为false，表示关闭自动确认机制，改为手动确认
channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> {});
 

@RabbitListener(queues = "queueName", acknowledgeMode = "MANUAL")
public void receiveMessage(Message message, Channel channel) throws IOException {
    String messageId = message.getMessageProperties().getMessageId();
    
    // 检查消息是否已消费
    if (messageAlreadyProcessed(messageId)) {
        // 消息已消费，确认消息并返回
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        return;
    }
    
    // 处理消息
    try {
        processMessage(message);
        // 消息处理成功，持久化消息ID并确认消息
        persistMessageId(messageId);
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    } catch (Exception e) {
        // 处理失败，可以选择重新入队或丢弃
        boolean requeue = shouldRequeue(message);
        channel.basicReject(message.getMessageProperties().getDeliveryTag(), requeue);
    }
}
 

void sendWithConfirm(AmqpTemplate amqpTemplate, Message message) throws IOException {
    ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
        if (!ack) {
            // 处理消息发送失败的逻辑
            // ...
        }
    };
    amqpTemplate.setConfirmCallback(confirmCallback);
    amqpTemplate.convertAndSend("exchangeName", "routingKey", message);
}

{
  "msg_id": "order_123",
  "parent_msg_id": "create_order_123",  // 依赖的上一个消息ID
  "version": 2,                        // 版本号（用于乐观锁）
  "status": "pending",                 // 状态机（pending/processing/done）
  "data": {"order_id": 123, "action": "deduct_stock"}
}

def consume_message(msg):
    # 1. 检查前置消息是否完成（如查询数据库或Redis）
    if not check_parent_msg_completed(msg["parent_msg_id"]):
        # 未完成则重新入队（或延迟重试）
        requeue_message(msg)
        return
 
    # 2. 乐观锁检查版本号
    if msg["version"] != get_current_version(msg["msg_id"]):
        return  # 已过期，丢弃
 
    # 3. 执行业务逻辑（如扣减库存）
    process_business(msg)
 
    # 4. 更新状态为完成
    update_msg_status(msg["msg_id"], "done")

// consumer处理成功后，通知broker删除队列中的消息，如果设置multiple=true，表示支持批量确认机制以减少网络流量
channel.basicAck(deliveryTag, multiple);
 
// 拒绝deliveryTag对应的消息，第二个参数是否requeue，true则重新入队列，否则丢弃或者进入死信队列，该方法reject后，该消费者还是会消费到该条被reject的消息
channel.basicReject(deliveryTag, requeue);
 
// 不确认 deliveryTag 对应的消息，第二个参数是否应用于多消息，第三个参数是否requeue，与basic.reject区别就是同时支持多个消息，可以nack该消费者先前接收未ack的所有消息。nack后的消息也会被自己消费到。
channel.basicNack(deliveryTag, multiple, requeue);
 
// 是否恢复消息到队列，参数是是否requeue，true则重新入队列，并且尽可能的将之前recover的消息投递给其他消费者消费，而不是自己再次消费。false则消息会重新被投递给自己。
channel.basicRecover(false);

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    listener:
      type: simple
      simple:
        default-requeue-rejected: false
        acknowledge-mode: manual

# Python pika 示例
channel.basic_consume(
    queue='order_queue',
    on_message_callback=callback,
    auto_ack=False  # 关闭自动ACK
)
 
def callback(ch, method, properties, body):
    try:
        process_order(body)  # 业务处理
        ch.basic_ack(delivery_tag=method.delivery_tag)  # 成功确认
    except Exception as e:
        logger.error(f"处理失败: {e}")
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)  # 不重试，进入DLQ

# 创建业务专用VHost
rabbitmqctl add_vhost order_vhost
rabbitmqctl set_permissions -p order_vhost app_user ".*" ".*" ".*"

params = pika.ConnectionParameters(
    host='rabbitmq.prod',
    virtual_host='order_vhost',  # 指定VHost
    credentials=pika.PlainCredentials('app_user', 'secure_password'),
    heartbeat=600,  # 心跳检测连接存活
    blocked_connection_timeout=30  # 网络阻塞超时
)

# 主队列声明时绑定DLX
channel.queue_declare(
    queue='order_queue',
    arguments={
        'x-dead-letter-exchange': 'dlx.order',
        'x-dead-letter-routing-key': 'failed.orders'
    }
)
 
# 死信队列声明
channel.exchange_declare('dlx.order', exchange_type='direct')
channel.queue_declare(queue='dead_letter.order')
channel.queue_bind('dead_letter.order', 'dlx.order', routing_key='failed.orders')

def process_dlx(ch, method, properties, body):
    logger.error(f"死信消息: {body}, 原因: {properties.headers.get('x-death')}")
    alert_admin(body)  # 通知管理员
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.queue_declare(
    queue='protected_queue',
    arguments={
        'x-max-length': 5000,          # 消息数限制
        'x-max-length-bytes': 10485760, # 10MB大小限制
        'x-overflow': 'reject-publish'  # 超限拒绝新消息
    }
)