RabbitMQ 消息可靠性

生产者 ──→ Exchange ──→ Queue ──→ 消费者
  │          │           │         │
  ├─ 发送失败 ├─ 路由失败 ├─ 持久化失败 ├─ 消费失败
  │          │           │         │
  └── 发送确认 └── 路由确认 └── 持久化  └── 手动确认

// 开启确认模式
channel.confirmSelect();
 
// 同步确认
channel.basicPublish(...);
channel.waitForConfirms();  // 阻塞等待确认
 
// 异步确认（推荐）
channel.addConfirmListener(new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        // 确认成功
    }
    
    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        // 确认失败，重发
    }
});

try {
    channel.txSelect();  // 开启事务
    channel.basicPublish(...);
    channel.txCommit();  // 提交事务
} catch (Exception e) {
    channel.txRollback();  // 回滚事务
}

// 批量发送后确认
for (int i = 0; i < 100; i++) {
    channel.basicPublish(...);
}
channel.waitForConfirms();  // 批量确认

// 队列持久化
channel.queueDeclare(
    "queue-name",
    true,   // durable: 是否持久化
    false,  // exclusive: 是否独占
    false,  // autoDelete: 是否自动删除
    null
);
 
// 消息持久化
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2)  // 2 表示持久化
    .build();
channel.basicPublish("", "queue-name", props, message.getBytes());

channel.exchangeDeclare(
    "exchange-name",
    "direct",
    true    // durable: 是否持久化
);

        ┌──────────────┐
        │   HA Proxy   │  负载均衡
        └──────┬───────┘
               │
    ┌──────────┼──────────┐
    ↓          ↓          ↓
┌────────┐ ┌────────┐ ┌────────┐
│ Node1  │ │ Node2  │ │ Node3  │
│ Master │ │ Slave  │ │ Slave  │
└────────┘ └────────┘ └────────┘
   同步镜像队列数据

# 管理界面策略配置
rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'

// 关闭自动确认
boolean autoAck = false;
channel.basicConsume("queue-name", autoAck, new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) {
        try {
            // 处理消息
            doWork(body);
            
            // 手动确认
            channel.basicAck(envelope.getDeliveryTag(), false);
        } catch (Exception e) {
            // 处理失败，拒绝消息
            // requeue = true: 重新入队
            // requeue = false: 丢弃或进入死信队列
            channel.basicNack(envelope.getDeliveryTag(), false, false);
        }
    }
});

// 限制未确认消息数量
channel.basicQos(10);  // 最多10条未确认消息

// 死信交换机
channel.exchangeDeclare("dlx.exchange", "direct");
 
// 业务队列绑定死信交换机
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx.exchange");
args.put("x-dead-letter-routing-key", "dlx.key");
 
channel.queueDeclare("business.queue", true, false, false, args);
 
// 死信队列
channel.queueDeclare("dlx.queue", true, false, false, null);
channel.queueBind("dlx.queue", "dlx.exchange", "dlx.key");

业务队列
   │
   ├─ 消息过期 ──→ 死信交换机 ──→ 死信队列
   ├─ 被拒绝   ──→
   └─ 队列满   ──→
                          │
                     后续处理
                     ├── 重试
                     ├── 告警
                     └── 人工处理

消费者处理消息 ──→ 确认失败（网络断开）
                      │
                      ↓
              消息重新投递 ──→ 重复消费

public void consume(Message message) {
    String messageId = message.getMessageId();
    
    // 检查是否已处理
    if (processedMessageDao.exists(messageId)) {
        return;  // 已处理，直接确认
    }
    
    // 处理消息（事务）
    transactionTemplate.execute(status -> {
        doWork(message);
        processedMessageDao.save(messageId);
        return null;
    });
}

public void consume(Message message) {
    String messageId = message.getMessageId();
    String key = "msg:processed:" + messageId;
    
    // SETNX 原子操作
    if (!redis.setNX(key, "1", 24 * 3600)) {
        return;  // 已处理
    }
    
    doWork(message);
}

// 订单状态机
public void updateOrderStatus(OrderMessage msg) {
    // UPDATE 只有当前状态符合条件才会更新
    int rows = orderMapper.updateStatus(
        msg.getOrderId(),
        "PAID",      // 目标状态
        "CREATED"    // 当前状态必须是 CREATED
    );
    
    if (rows == 0) {
        // 状态已更新，幂等返回
        return;
    }
}

队列: [msg1, msg2, msg3]
消费者A: 处理 msg1 (慢)
消费者B: 处理 msg2 (快) ──→ msg2 先完成
结果: msg2 在 msg1 之前处理完成

// 限制只有一个消费者
channel.basicQos(1);  // 每次只取一条

同一订单的消息路由到同一队列

Exchange
   │
   ├── 订单1 ──→ Queue-1 ──→ Consumer-1
   ├── 订单2 ──→ Queue-2 ──→ Consumer-2
   └── 订单3 ──→ Queue-3 ──→ Consumer-3

生产者 ──→ 发送确认（Confirm 模式）
   ↓
Broker  ──→ 持久化（Exchange + Queue + Message）
   ↓
消费者 ──→ 手动确认（basicAck）

消息 ──→ 延迟队列（TTL）──→ 死信交换机 ──→ 业务队列

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

// 消息头记录重试次数
public void consume(Message message) {
    int retryCount = getRetryCount(message);
    
    try {
        doWork(message);
        channel.basicAck(...);
    } catch (Exception e) {
        if (retryCount < MAX_RETRY) {
            // 重试：重新入队
            channel.basicNack(..., true);
        } else {
            // 超过重试次数：进入死信队列
            channel.basicNack(..., false);
        }
    }
}

// 1. 开启 Confirm 模式
channel.confirmSelect();
 
// 2. 异步确认
channel.addConfirmListener(...);
 
// 3. 消息持久化
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2)
    .messageId(UUID.randomUUID().toString())  // 幂等ID
    .build();
 
// 4. 发送超时和重试

// 1. 手动确认
boolean autoAck = false;
 
// 2. 预取限制
channel.basicQos(10);
 
// 3. 幂等处理
// 4. 异常处理和重试
// 5. 死信队列兜底