// 生产者
DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();

Message msg = new Message("order_topic", "order_created",
    ("订单创建事件:" + orderId).getBytes("UTF-8"));
SendResult result = producer.send(msg);
System.out.println("发送结果：" + result.getSendStatus());
producer.shutdown();

// 消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.subscribe("order_topic", "*");
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (Message msg : msgs) {
        System.out.println("收到消息：" + new String(msg.getBody()));
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();

生产者（瞬时高并发）
        │
        ▼
┌─────────────────┐
│   MQ Broker     │  ← 消息存储层，负责缓冲
│  (持久化存储)    │
└─────────────────┘
        │
        ▼
消费者（匀速或限速消费）

// 消费者端设置拉取速率，实现匀速消费
// Kafka 示例：通过 max.poll.records 控制每次拉取量
Properties props = new Properties();
props.put("max.poll.records", 100);       // 每次最多拉取 100 条
props.put("max.poll.interval.ms", 300000); // 拉取间隔最大 5 分钟

// RocketMQ 示例：通过线程池控制消费速率
consumer.registerMessageListener(new MessageListenerConcurrently() {
    private final Semaphore semaphore = new Semaphore(10); // 最多 10 个并发

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                    ConsumeConcurrentlyStatus context) {
        for (MessageExt msg : msgs) {
            semaphore.acquire();
            try {
                processMessage(msg);
            } finally {
                semaphore.release();
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

用户下单 → 创建订单（核心） → 发送短信通知（异步）
                              → 扣减积分（异步）
                              → 推送物流（异步）
                              → 统计埋点（异步）

MySQL Binlog → Canal → MQ → Elasticsearch（更新索引）
                         → Redis（刷新缓存）
                         → Hive（数据仓库）

订单服务（本地事务：创建订单 + 写消息表）
    ↓
MQ 事务消息（发送"订单已创建"）
    ↓
支付服务（消费消息，执行支付逻辑）

生产者 ──发送──▶ Broker ──推送/拉取──▶ 消费者
  │                │                      │
  ▼                ▼                      ▼
 ① 发送失败     ② 存储丢失             ③ 处理失败

# 生产者配置
acks=all                    # 等待 ISR 中所有副本确认
retries=3                   # 失败重试次数
enable.idempotence=true     # 幂等生产者，避免重试导致重复
max.in.flight.requests.per.connection=5  # 限制在途请求数

// 同步发送，等待 Broker 响应
SendResult result = producer.send(msg);
if (result.getSendStatus() == SendStatus.SEND_OK) {
    // 发送成功
} else {
    // 处理失败
}

// 先处理业务，再确认
consumer.registerMessageListener((msgs, context) -> {
    try {
        for (Message msg : msgs) {
            processMessage(msg);  // 业务逻辑
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  // 确认
    } catch (Exception e) {
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;  // 不确认，稍后重试
    }
});

// 1. 生产者：同步发送 + 重试 + 幂等
DefaultMQProducer producer = new DefaultMQProducer("group");
producer.setSendMsgTimeout(10000);
producer.start();

Message msg = new Message("topic", "tag", "body".getBytes());
SendResult result = producer.send(msg);  // 同步发送
assert result.getSendStatus() == SendStatus.SEND_OK;

// 2. Broker：同步刷盘（Broker 配置）
// broker.conf: flushDiskType=SYNC_FLUSH

// 3. 消费者：手动确认
consumer.registerMessageListener((msgs, context) -> {
    MessageExt msg = msgs.get(0);
    try {
        // 处理业务
        handleBusiness(msg);
        // 业务成功才确认
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    } catch (Exception e) {
        // 记录日志，稍后重试
        log.error("消息处理失败: {}", msg, e);
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
});

CREATE TABLE msg_deduplicate (
    msg_id VARCHAR(128) PRIMARY KEY,
    consume_status TINYINT DEFAULT 0,
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP
);

public void consume(Message msg) {
    String msgId = msg.getMsgId();
    // 1. 查询消息是否已处理
    MsgDeduplicate record = deduplicateMapper.selectById(msgId);
    if (record != null && record.getConsumeStatus() == 1) {
        return;  // 已处理，直接返回
    }
    // 2. 开始处理（使用事务保证原子性）
    transactionTemplate.execute(status -> {
        // 先标记消息为处理中
        deduplicateMapper.insertIgnore(new MsgDeduplicate(msgId, 0));
        // 执行业务逻辑
        doBusiness(msg);
        // 标记处理完成
        deduplicateMapper.updateStatus(msgId, 1);
        return null;
    });
}

public void consume(Message msg) {
    String dedupKey = "msg:consumed:" + msg.getMsgId();
    // SETNX 原子操作，key 不存在才设置成功
    Boolean success = redisTemplate.opsForValue()
        .setIfAbsent(dedupKey, "1", 24, TimeUnit.HOURS);
    if (Boolean.FALSE.equals(success)) {
        return;  // 已处理
    }
    // 执行业务逻辑
    doBusiness(msg);
}

public void consume(Message msg) {
    String orderId = msg.getOrderId();
    Order order = orderMapper.selectById(orderId);
    // 只有待支付状态才能处理
    if (order.getStatus() != OrderStatus.PENDING_PAYMENT) {
        return;  // 已处理过
    }
    // 更新状态（CAS 语义）
    int updated = orderMapper.updateStatus(orderId,
        OrderStatus.PENDING_PAYMENT, OrderStatus.PAID);
    if (updated == 0) {
        return;  // 已被其他请求处理
    }
    // 执行业务逻辑
    doBusiness(msg);
}

UPDATE account SET balance = balance - #{amount}, version = version + 1
WHERE id = #{id} AND version = #{version}

@Component
public class IdempotentConsumer {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String DEDUP_PREFIX = "mq:consumed:";
    private static final int DEDUP_EXPIRE_HOURS = 24;

    public ConsumeConcurrentlyStatus consume(MessageExt msg) {
        String dedupKey = DEDUP_PREFIX + msg.getMsgId();

        // 1. 先做幂等检查
        Boolean isFirst = redisTemplate.opsForValue()
            .setIfAbsent(dedupKey, "1", DEDUP_EXPIRE_HOURS, TimeUnit.HOURS);
        if (Boolean.FALSE.equals(isFirst)) {
            log.warn("消息重复消费，msgId={}", msg.getMsgId());
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        // 2. 执行业务逻辑
        try {
            doBusiness(msg);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            // 业务失败时删除去重标记，允许重试
            redisTemplate.delete(dedupKey);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}

// 生产者：指定分区键（如订单 ID）
ProducerRecord<String, String> record = new ProducerRecord<>(
    "order_topic",
    orderId,  // 分区键，相同 orderId 的消息会路由到同一分区
    message
);
producer.send(record);

// 消费者：消费者组内每个分区由一个消费者实例消费
// 配置：消费顺序保证在分区级别

// 1. 全局有序（一个 Topic 只有一个队列）
// 设置 Topic 为全局有序

// 2. 分区有序（同一业务 key 路由到同一队列）
DefaultMQProducer producer = new DefaultMQProducer("group");
// 使用选择器策略
producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        String orderId = (String) arg;
        int index = Math.abs(orderId.hashCode()) % mqs.size();
        return mqs.get(index);
    }
}, orderId);

consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        processMessage(msg);
    }
    return ConsumeOrderlyStatus.SUCCESS;
});

// 声明持久化、非排他、自动删除的队列
// 每个业务对象对应一个队列
String queueName = "order_" + orderId;
channel.queueDeclare(queueName, true, false, false, null);
channel.basicConsume(queueName, false, deliverCallback, consumerTag -> {});

// 生产者：订单 ID 作为分区键
public void sendOrderEvent(OrderEvent event) {
    String key = event.getOrderId();
    String value = JSON.toJSONString(event);

    // Kafka
    producer.send(new ProducerRecord<>("order_event", key, value));

    // RocketMQ
    Message msg = new Message("order_event", value.getBytes());
    producer.send(msg, (mqs, msg1, arg) -> {
        int idx = Math.abs(arg.hashCode()) % mqs.size();
        return mqs.get(idx);
    }, key);
}

// 消费者：确保同一 orderId 的事件在同一分区顺序消费
// 不需要额外代码，只要分区键相同即可

第一步：评估积压规模和增长速率
    │
    ▼
第二步：紧急扩容消费者
    │
    ├── 扩容有效 → 持续消费直到积压清除 → 排查根因
    │
    ├── 扩容无效 → 优化消费者处理逻辑 / 临时降级
    │
    └── 极端情况 → 丢弃非核心消息

// 优化前：单条处理
for (Message msg : messages) {
    processOne(msg);  // 每条消息都要调用一次数据库
}

// 优化后：批量处理
batchProcess(messages);  // 批量写入/更新数据库

@Service
public class MessageConsumerService {

    @Value("${consumer.batch-size:100}")
    private int batchSize;

    private final AtomicBoolean emergencyMode = new AtomicBoolean(false);

    public ConsumeConcurrentlyStatus consume(List<MessageExt> msgs) {
        List<MessageExt> toProcess = msgs;

        // 紧急模式：只处理核心消息
        if (emergencyMode.get()) {
            toProcess = msgs.stream()
                .filter(m -> isCoreMessage(m))
                .collect(Collectors.toList());
            if (toProcess.isEmpty()) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        }

        // 批量处理
        try {
            batchProcess(toProcess);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error("消费失败", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

    private boolean isCoreMessage(MessageExt msg) {
        String tag = msg.getTags();
        return "order_created".equals(tag) || "payment_success".equals(tag);
    }

    // 紧急模式开关（通过管理接口触发）
    @MonitorOperation
    public void toggleEmergency(boolean on) {
        emergencyMode.set(on);
    }
}

# 积压监控规则
alert:
  - name: mq_backlog_critical
    condition: backlog > 1000000
    level: critical
    notification: [sms, phone]
  - name: mq_backlog_warning
    condition: backlog > 100000
    level: warning
    notification: [email]

Topic: order_topic
    │
    ├── Partition 0 (Broker 1) → Log: [msg1, msg2, msg3, ...]
    ├── Partition 1 (Broker 2) → Log: [msg4, msg5, msg6, ...]
    └── Partition 2 (Broker 3) → Log: [msg7, msg8, msg9, ...]

// 生产者写入时的分区选择策略
// 1. 指定了 partition → 直接写入
// 2. 有 key → hash(key) % numPartitions
// 3. 无 key → 轮询（round-robin）或粘性分区（Kafka 2.4+）

Producer → RecordAccumulator → Sender Network Thread
    │
    ▼
按分区聚合 → 批量发送到对应 Broker
    │
    ▼
Broker Leader → 写入本地 Log → 同步到 Follower → 返回 ACK

消费者组：
    Consumer 0 → 消费 Partition 0
    Consumer 1 → 消费 Partition 1
    Consumer 2 → 消费 Partition 2

# 创建 Topic：3 个分区，2 个副本
kafka-topics.sh --create \
    --bootstrap-server localhost:9092 \
    --topic order_topic \
    --partitions 3 \
    --replication-factor 2

// 按订单 ID 分区，保证同一订单的消息在同一分区
String orderId = "ORDER_20240101_001";
ProducerRecord<String, String> record = new ProducerRecord<>(
    "order_topic",
    orderId,           // key：分区键
    "{\"orderId\":\"...\"}"  // value：消息体
);
producer.send(record, (metadata, exception) -> {
    if (exception == null) {
        System.out.println("发送成功，分区：" + metadata.partition());
    }
});

# 增加分区数（只能增加，不能减少）
kafka-topics.sh --alter \
    --bootstrap-server localhost:9092 \
    --topic order_topic \
    --partitions 6

Topic: order_topic (3 partitions)
    │
    ▼
消费者组: order_consumer_group
    ├── Consumer A → Partition 0, Partition 1
    └── Consumer B → Partition 2

如果 Consumer A 挂了 → Partition 0, 1 转移给 Consumer B

1. 消费者启动 → 加入消费者组
2. GroupCoordinator 协调分配
3. 触发 Rebalance 过程
   │
   ├── 第一阶段（JoinGroup）：所有消费者声明自己想消费的分区
   ├── 第二阶段（SyncGroup）：由 Leader 消费者制定分配方案，同步给其他消费者
   └── 第三阶段（稳定消费）：按分配方案消费各自的分区

__consumer_offsets (Kafka 内部 Topic)
    │
    ├── Group ID + Topic + Partition → Offset
    ├── order_group + order_topic + 0 → 12345
    ├── order_group + order_topic + 1 → 12340
    └── ...

// 配置消费者属性
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order_consumer_group");  // 消费者组 ID
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("enable.auto.commit", "false");  // 关闭自动提交
props.put("max.poll.records", 100);         // 每次最多拉取 100 条

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("order_topic"));

// 主消费循环
try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("消费到消息：partition=%d, offset=%d, key=%s, value=%s%n",
                record.partition(), record.offset(), record.key(), record.value());
            // 处理业务逻辑
            processBusiness(record);
        }
        // 手动提交 Offset
        consumer.commitSync();
    }
} finally {
    consumer.close();
}

# 增加心跳超时和会话超时，避免误判为宕机
session.timeout.ms=30000        # 30 秒
heartbeat.interval.ms=10000      # 10 秒（通常是 session.timeout.ms 的 1/3）
max.poll.interval.ms=300000      # 5 分钟（处理业务的最大间隔）

# 使用 Incremental Cooperative Rebalance（Kafka 2.4+）
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

// 生产者发送时的关键配置
Properties props = new Properties();
props.put("enable.idempotence", "true");           // 启用幂等生产者
props.put("max.in.flight.requests.per.connection", "1");  // 仅允许 1 个在途请求
props.put("acks", "all");                          // 等待所有 ISR 副本确认

// 关键：分区键选择
// 订单场景：相同 orderId 的消息路由到同一分区
String orderId = "ORDER_20240101_001";
producer.send(new ProducerRecord<>("order_topic", orderId, value));

// Kafka 路由规则：hash(key) % numPartitions
// 相同 key → 相同 partition → 分区内有序

消费者组内分区分配：
    Consumer A → Partition 0（订单 0-99）
    Consumer B → Partition 1（订单 100-199）

Partition 0 的消息在 Consumer A 上严格有序消费。

// 生产者配置
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "localhost:9092");
producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
producerProps.put("enable.idempotence", "true");
producerProps.put("max.in.flight.requests.per.connection", "1");
producerProps.put("acks", "all");

KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

// 发送消息：以订单 ID 为 key，保证同一订单的消息在同一分区
String orderId = "ORDER_20240101_001";
String message = "{\"event\":\"created\",\"orderId\":\"" + orderId + "\"}";
producer.send(new ProducerRecord<>("order_topic", orderId, message));

// 消费者配置
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "localhost:9092");
consumerProps.put("group.id", "order_group");
consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
consumerProps.put("enable.auto.commit", "false");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("order_topic"));

// 消费：每个分区独立有序
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> record : records) {
        // 同一分区内的消息有序到达
        processOrderEvent(record.key(), record.value());
    }
    consumer.commitSync();
}

Producer → Topic → Partition 0 → Broker 1 (Leader) → ISR 副本 → 消费者组
                → Partition 1 → Broker 2 (Leader) → ISR 副本 → 消费者组
                → Partition 2 → Broker 3 (Leader) → ISR 副本 → 消费者组

Producer → Topic → Queue 0 → CommitLog (全局追加)
                → Queue 1 → CommitLog (全局追加)
                → Queue 2 → CommitLog (全局追加)

// 适合日志收集场景
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "...");
props.put("value.serializer", "...");

// 日志消息直接发送，无需复杂路由
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("log_topic", logMessage));

// 适合电商订单场景
DefaultMQProducer producer = new DefaultMQProducer("order_group");
producer.setNamesrvAddr("localhost:9876");
producer.start();

// 1. 普通消息
Message msg = new Message("order_topic", "order_created", body);
producer.send(msg);

// 2. 延迟消息（30 分钟后消费）
Message delayMsg = new Message("order_topic", body);
delayMsg.setDelayLevel(MessageConst.DELAY_LEVEL_30M);
producer.send(delayMsg);

// 3. 事务消息
TransactionListener listener = new OrderTransactionListener();
producer.sendMessageInTransaction(msg, arg);

下单流程：
1. 创建订单（本地事务）
2. 发送订单创建事件（事务消息）
3. 扣减库存（消费者处理）
4. 发送通知（消费者处理）

// 事务消息：保证步骤 1 和 2 的原子性
TransactionSendResult result = producer.sendMessageInTransaction(
    msg, new OrderTransactionListener(), orderParam);

下单 30 分钟未支付 → 自动关闭订单

Message msg = new Message("order_topic", "close_order", orderId.getBytes());
msg.setDelayLevel(MessageConst.DELAY_LEVEL_30M);  // 30 分钟延迟
producer.send(msg);

订单状态：创建 → 支付 → 发货 → 完成（必须严格有序）

// 创建全局有序 Topic（只有一个队列）
mqAdmin.updateTopicConfig("order_topic", ..., 1);  // 1 个队列，全局有序

// 使用顺序消息发送
SendResult result = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        return mqs.get(0);  // 选择唯一的队列
    }
}, orderId);

// RocketMQ 完整使用示例
DefaultMQProducer producer = new DefaultMQProducer("order_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();

// 1. 普通消息
Message normalMsg = new Message("order_topic", "order_created",
    JSON.toJSONString(orderEvent).getBytes());
producer.send(normalMsg);

// 2. 延迟消息
Message delayMsg = new Message("order_topic", "close_order",
    orderId.getBytes());
delayMsg.setDelayLevel(MessageConst.DELAY_LEVEL_30M);
producer.send(delayMsg);

// 3. 事务消息
Message transactionMsg = new Message("order_topic", "order_created",
    JSON.toJSONString(orderEvent).getBytes());
TransactionSendResult txResult = producer.sendMessageInTransaction(
    transactionMsg, new OrderTransactionListener(), orderParam);

// 4. 顺序消息（分区有序）
Message seqMsg = new Message("order_topic", "order_created",
    JSON.toJSONString(orderEvent).getBytes());
producer.send(seqMsg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        int idx = Math.abs(arg.hashCode()) % mqs.size();
        return mqs.get(idx);
    }
}, orderId);

1s, 5s, 10s, 30s,
1m, 2m, 5m, 10m, 30m,
1h, 2h, 6h, 12h, 1d, 2d, 3d, 7d

生产者发送延迟消息
        │
        ▼
┌─────────────────────────────┐
│ 发送到 RMQ_SYS_SCHEDULE_TOPIC│
│ (特殊的调度 Topic)            │
│ 按 delayLevel 路由到对应 Queue│
│ Queue 索引 = delayLevel - 2  │
└─────────────────────────────┘
        │
        ▼
ScheduleMessageService 定时轮询（1秒一次）
        │
        ├── 检查消息的投递时间是否到达
        │
        ├── 未到达 → 跳过
        │
        └── 已到达 → 重新投递到原始 Topic 的原始 Queue
                              │
                              ▼
                     原始消费者正常消费

                    时间轮（TimerWheel）
            ┌─────────────────────┐
            │  0  1  2  3 ...  59 │  ← 60 个槽位
            │  ↑                   │
            │  指针每 1 秒移动一格   │
            └─────────────────────┘
                    │
                    ▼
        每个槽位维护一个延迟消息列表
                    │
                    ▼
        指针到达某槽位 → 投递该槽位的所有消息
                    │
                    ▼
        消息被投递到原始 Topic

// RocketMQ 4.x：使用固定延迟等级
DefaultMQProducer producer = new DefaultMQProducer("order_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();

// 创建延迟消息：30 分钟后投递
Message msg = new Message("order_topic", "close_order",
    orderId.getBytes("UTF-8"));
msg.setDelayLevel(MessageConst.DELAY_LEVEL_30M);

SendResult result = producer.send(msg);
System.out.println("延迟消息 ID：" + result.getMsgId());

// RocketMQ 5.x：使用任意延迟时间
// 使用新的 API 设置精确延迟
org.apache.rocketmq.client.apis.message.Message newMsg =
    org.apache.rocketmq.client.apis.message.Message.builder()
    .setTopic("order_topic")
    .setBody(orderId.getBytes())
    .setDelayDuration(java.time.Duration.ofMinutes(30))  // 任意延迟时间
    .build();

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_group");
consumer.subscribe("order_topic", "close_order");
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        String orderId = new String(msg.getBody());
        closeOrder(orderId);  // 30 分钟后执行
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();

方案一：先更新数据库，再发送消息
    → 如果数据库成功但消息发送失败 → 数据不一致

方案二：先发送消息，再更新数据库
    → 如果消息发送成功但数据库更新失败 → 数据不一致

方案三：本地消息表
    → 需要在同一个事务中写消息表，增加了数据库设计复杂度

                    ┌─────────────────┐
                    │   生产者        │
                    └────────┬────────┘
                             │
        ① 发送半消息（对消费者不可见）
                             │
                             ▼
                    ┌─────────────────┐
                    │   Broker        │
                    │ 存储半消息       │
                    └────────┬────────┘
                             │
        ② 执行本地事务（如更新数据库）
                             │
                    ┌────────┴────────┐
                    ▼                 ▼
              本地事务成功        本地事务失败
                    │                 │
        ③a. Commit 消息        ③b. Rollback 消息
            (对消费者可见)        (消息被丢弃)
                    │                 │
                    ▼                 ▼
              消费者正常消费      消息不被投递

        如果 ③ 步骤超时：
        Broker 发起事务回查 → 生产者返回本地事务状态
            ├── COMMIT_MESSAGE → 提交消息
            ├── ROLLBACK_MESSAGE → 回滚消息
            └── UNKNOW → 稍后再次回查

@Transactional
public void executeLocal事务(Order order) {
    orderMapper.insert(order);  // 插入订单记录
    // ... 其他数据库操作
}

public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    String orderId = new String(msg.getBody());
    // 查询本地事务状态
    Order order = orderMapper.selectById(orderId);
    if (order != null) {
        return LocalTransactionState.COMMIT_MESSAGE;  // 订单存在 → 提交
    } else {
        return LocalTransactionState.ROLLBACK_MESSAGE;  // 订单不存在 → 回滚
    }
}

// 1. 实现事务监听器
public class OrderTransactionListener implements TransactionListener {

    @Autowired
    private OrderService orderService;

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        Order order = (Order) arg;
        try {
            // 执行本地事务
            orderService.createOrder(order);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(Message msg) {
        String orderId = new String(msg.getBody());
        // 回查本地事务状态
        Order order = orderService.getOrder(orderId);
        if (order != null) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
}

// 2. 发送事务消息
DefaultMQProducer producer = new DefaultMQProducer("order_group");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.setTransactionListener(new OrderTransactionListener());
producer.start();

Order order = new Order("ORDER_001", "user_001", ...);
Message msg = new Message("order_topic", "order_created",
    order.getOrderId().getBytes());

// 发送事务消息（半消息 + 本地事务 + 确认）
TransactionSendResult result = producer.sendMessageInTransaction(msg, order);

Broker 1 (Master)          Broker 2 (Mirror)
    │                           │
    ├── Queue: order_queue      ├── Queue: order_queue (mirror)
    │   ├── msg1 ✓              │   ├── msg1 ✓ (同步)
    │   ├── msg2 ✓              │   ├── msg2 ✓ (同步)
    │   └── msg3 ✓              │   └── msg3 ✓ (同步)
    │                           │
    └── 如果 Broker 1 宕机 ──────→  Broker 2 提升为 Master

# 创建镜像策略：将 order_queue 镜像到所有节点
rabbitmqctl set_policy ha-all "^order" \
    '{"ha-mode":"all","ha-sync-mode":"automatic"}'

生产者 → Broker 1 (Master Queue)
    │
    ├── 写入主队列
    │
    ├── 同步到 Broker 2 (Mirror Queue)
    │   → 同步完成后才返回 ACK
    │
    └── 返回确认给生产者

1. Broker 1 (Master) 宕机
    │
    ▼
2. RabbitMQ 检测到主队列不可用
    │
    ▼
3. 选择一个镜像节点（如 Broker 2）
    │
    ▼
4. 将镜像节点提升为新的 Master
    │
    ▼
5. 生产者/消费者重连到新的 Master

// 1. 声明镜像队列
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

// 声明队列（镜像策略通过 RabbitMQ 管理端或命令行配置）
channel.queueDeclare("order_queue", true, false, false, null);

// 2. 生产消息
channel.basicPublish("", "order_queue", null, "message".getBytes());

// 3. 消费消息（自动故障转移）
 DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    System.out.println("收到消息：" + message);
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
};

channel.basicConsume("order_queue", false, deliverCallback, consumerTag -> {});

// 使用 Quorum Queue 替代传统镜像队列
Map<String, Object> args = new HashMap<>();
args.put("x-queue-type", "quorum");  // 基于 Raft 协议

channel.queueDeclare("order_queue", true, false, false, args);

生产者 ──发送──▶ Broker ──推送──▶ 消费者
  │                │              │
  ▼                ▼              ▼
 ① 发送失败     ② 存储丢失    ③ 处理失败

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

// 开启发布确认
channel.confirmSelect();

// 声明持久化交换机和队列
channel.exchangeDeclare("order_exchange", "direct", true);
channel.queueDeclare("order_queue", true, false, false, null);
channel.queueBind("order_queue", "order_exchange", "order_routing");

// 发布消息（mandatory 确保路由失败时返回）
channel.basicPublish("order_exchange", "order_routing", true,
    MessageProperties.PERSISTENT_TEXT_PLAIN,
    "order_created".getBytes());

// 确认消息到达 Broker
if (channel.waitForConfirms()) {
    System.out.println("消息已到达 Broker");
} else {
    System.out.println("消息可能丢失");
}

// 声明持久化队列（durable=true）
channel.queueDeclare("order_queue",
    true,   // durable：持久化
    false,  // exclusive：排他
    false,  // autoDelete：自动删除
    null);

// 配合镜像队列或 Quorum Queue 实现高可用
// 镜像队列配置：rabbitmqctl set_policy ha-order "^order" '{"ha-mode":"all"}'
// Quorum Queue：args.put("x-queue-type", "quorum")

// 消费者：手动 ACK
boolean autoAck = false;  // 关闭自动 ACK
channel.basicConsume("order_queue", autoAck, deliverCallback, consumerTag -> {});

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    long deliveryTag = delivery.getEnvelope().getDeliveryTag();

    try {
        // 1. 幂等检查
        String msgId = delivery.getProperties().getMessageId();
        Boolean isFirst = redisTemplate.opsForValue()
            .setIfAbsent("consumed:" + msgId, "1", 24, TimeUnit.HOURS);
        if (Boolean.FALSE.equals(isFirst)) {
            // 重复消息，直接 ACK
            channel.basicAck(deliveryTag, false);
            return;
        }

        // 2. 处理业务
        processMessage(message);

        // 3. 手动 ACK（处理完成后）
        channel.basicAck(deliveryTag, false);
    } catch (Exception e) {
        // 4. 处理失败，不 ACK，消息会重新入队
        // 可以设置重试次数或转入死信队列
        channel.basicNack(deliveryTag, false, true);  // 重新入队
    }
};

// 声明死信交换机和队列
channel.exchangeDeclare("dlx_exchange", "direct", true);
channel.queueDeclare("dlx_queue", true, false, false, null);
channel.queueBind("dlx_queue", "dlx_exchange", "dlx_routing");

// 业务队列配置死信路由
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx_exchange");
args.put("x-dead-letter-routing-key", "dlx_routing");
args.put("x-message-ttl", 60000);  // TTL 60 秒

channel.queueDeclare("order_queue", true, false, false, args);

@Component
public class ReliableRabbitMQConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setPublisherConfirmType(ConnectionFactory.ConfirmType.CORRELATED);
        factory.setPublisherReturns(true);
        return factory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory factory) {
        RabbitTemplate template = new RabbitTemplate(factory);
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.error("消息发送失败：{}", cause);
                // 补偿：写入本地消息表
            }
        });
        template.setReturnsCallback(returned -> {
            log.error("消息路由失败：{}", returned.getMessage());
        });
        return template;
    }
}

Broker 1 (Master)            Broker 2 (Mirror)
    │                           │
    ├── order_queue             ├── order_queue (mirror)
    │   ├── msg1 ✓              │   ├── msg1 ✓ (同步)
    │   ├── msg2 ✓              │   ├── msg2 ✓ (同步)
    │   └── msg3 ✓              │   └── msg3 ✓ (同步)
    │                           │
    └── Broker 1 宕机 ──────────→ Broker 2 自动提升为 Master

# 查看当前策略
rabbitmqctl list_policies

# 创建镜像策略：以 "order" 开头的队列镜像到所有节点
rabbitmqctl set_policy ha-order "^order" \
    '{"ha-mode":"all","ha-sync-mode":"automatic"}' \
    --priority 0

# 创建镜像策略：镜像到指定数量的节点
rabbitmqctl set_policy ha-order-count "^order" \
    '{"ha-mode":"exactly","ha-params":2,"ha-sync-mode":"automatic"}'

生产者 → Broker 1 (Master Queue)
    │
    ├── 1. 消息写入本地存储
    │
    ├── 2. 通过 AMQP Basic.Publish 命令同步到 Mirror
    │      Broker 2 (Mirror Queue) 接收并存储
    │
    ├── 3. 所有 Mirror 确认同步完成
    │
    └── 4. 返回 Basic.Ack 给生产者

1. Broker 1 (Master) 宕机
    │
    ▼
2. 集群检测到 Master 不可用（心跳超时）
    │
    ▼
3. 根据策略选择一个 Mirror 节点（如 Broker 2）
    │
    ▼
4. 将 Broker 2 提升为新的 Master
    │
    ▼
5. 通知生产者/消费者重连到新 Master
    │
    ▼
6. 新 Master 继续接收和处理消息

// 生产者
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("rabbitmq-host");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

// 声明持久化队列
channel.queueDeclare("order_queue", true, false, false, null);

// 发送消息
byte[] messageBody = "order_created:{...}".getBytes();
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2)  // 持久化消息
    .messageId(UUID.randomUUID().toString())
    .build();

channel.basicPublish("", "order_queue", props, messageBody);

// 消费者
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    long tag = delivery.getEnvelope().getDeliveryTag();

    try {
        processMessage(message);
        channel.basicAck(tag, false);
    } catch (Exception e) {
        channel.basicNack(tag, false, true);
    }
};

channel.basicConsume("order_queue", false, deliverCallback, consumerTag -> {});

// 使用 Quorum Queue 替代镜像队列
Map<String, Object> args = new HashMap<>();
args.put("x-queue-type", "quorum");  // 基于 Raft 协议的仲裁队列

channel.queueDeclare("order_queue", true, false, false, args);

生产者 ──发送──▶ Broker ──推送──▶ 消费者
  │                │              │
  ▼                ▼              ▼
 ① 发送失败     ② 存储丢失    ③ 处理失败

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");

// 方式一：Channel 级别的 confirmSelect（传统方式）
// channel.confirmSelect();
// channel.basicPublish(...);
// channel.waitForConfirms();

// 方式二：Connection 级别的 Correlated Confirm（推荐）
factory.setPublisherConfirmType(ConnectionFactory.ConfirmType.CORRELATED);
factory.setPublisherReturns(true);

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

channel.confirmSelect();

// 发布确认回调
ConfirmListener confirmListener = new ConfirmListener() {
    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        // Broker 已确认收到消息
        System.out.println("消息已确认: tag=" + deliveryTag);
    }

    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        // Broker 拒绝了消息
        System.out.println("消息被拒绝: tag=" + deliveryTag);
        // 补偿处理：写入本地消息表或重试
    }
};
channel.addConfirmListener(confirmListener);

// Return 回调（mandatory=true 时消息无法路由触发）
ReturnListener returnListener = (replyText, replyCode, exchange, routingKey, properties, body) -> {
    System.out.println("消息路由失败: " + replyText);
    // 补偿处理
};
channel.addReturnListener(returnListener);

// 发布持久化消息
channel.basicPublish("order_exchange", "order_created",
    true,  // mandatory：消息无法路由时通过 Return 回调返回
    MessageProperties.PERSISTENT_TEXT_PLAIN,
    "order_created:{...}".getBytes());

// 同步等待确认
if (channel.waitForConfirms(5000)) {
    System.out.println("消息已成功发送到 Broker");
}

@Configuration
public class RabbitMQConfig {

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);

        // 发布确认回调
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.error("消息发送失败：{}", cause);
                // 补偿逻辑
            }
        });

        // Return 回调
        template.setReturnsCallback(returned -> {
            log.error("消息路由失败：{}", returned.getMessage());
        });

        // 使用 RabbitTemplate 时自动开启 Publisher Confirm
        return template;
    }
}

// durable=true：队列持久化
// MessageProperties.PERSISTENT_TEXT_PLAIN：消息持久化
channel.exchangeDeclare("order_exchange", "direct", true);  // 持久化交换机
channel.queueDeclare("order_queue", true, false, false, null);  // 持久化队列
channel.queueBind("order_queue", "order_exchange", "order_created");

// 方式一：镜像队列
rabbitmqctl set_policy ha-order "^order" '{"ha-mode":"all"}'

// 方式二：Quorum Queue（RabbitMQ 4.x 推荐）
Map<String, Object> args = new HashMap<>();
args.put("x-queue-type", "quorum");
channel.queueDeclare("order_queue", true, false, false, args);

// autoAck = false：关闭自动确认
boolean autoAck = false;
channel.basicConsume("order_queue", autoAck, deliverCallback, consumerTag -> {});

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    long deliveryTag = delivery.getEnvelope().getDeliveryTag();
    String message = new String(delivery.getBody(), "UTF-8");

    try {
        // 处理业务逻辑
        processMessage(message);

        // 处理成功后手动 ACK
        channel.basicAck(deliveryTag, false);  // false = 不批量确认
    } catch (Exception e) {
        // 处理失败，不 ACK，消息会重新入队
        // 可设置重试次数或转入死信队列
        channel.basicNack(deliveryTag, false, true);  // true = 重新入队
    }
};

@Component
public class IdempotentMessageConsumer {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String DEDUP_PREFIX = "mq:consumed:";
    private static final int DEDUP_HOURS = 24;

    @RabbitListener(queues = "order_queue")
    public void handleMessage(String message, Channel channel,
                               Message messageObj, Consumer metadata) throws IOException {
        long deliveryTag = messageObj.getMessageProperties().getDeliveryTag();

        // 1. 幂等检查
        String msgId = messageObj.getMessageProperties().getMessageId();
        Boolean isFirst = redisTemplate.opsForValue()
            .setIfAbsent(DEDUP_PREFIX + msgId, "1", DEDUP_HOURS, TimeUnit.HOURS);

        if (Boolean.FALSE.equals(isFirst)) {
            // 重复消息，直接 ACK
            channel.basicAck(deliveryTag, false);
            return;
        }

        try {
            // 2. 处理业务
            OrderEvent event = JSON.parseObject(message, OrderEvent.class);
            handleOrderCreated(event);

            // 3. 成功后 ACK
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            // 4. 失败时删除去重标记，允许重试
            redisTemplate.delete(DEDUP_PREFIX + msgId);
            channel.basicNack(deliveryTag, false, true);
        }
    }

    // 业务逻辑
    private void handleOrderCreated(OrderEvent event) {
        // 数据库唯一键去重（兜底）
        try {
            orderMapper.insert(event.toEntity());
        } catch (DuplicateKeyException e) {
            // 已处理过，直接返回
            return;
        }
        // 执行业务
        inventoryService.deduct(event.getOrderId(), event.getProductId());
    }
}

@Configuration
public class DeadLetterConfig {

    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange("dlx_exchange", true, false);
    }

    @Bean
    public Queue dlxQueue() {
        return new Queue("dlx_queue", true);
    }

    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("dlx_routing");
    }

    // 业务队列配置死信路由
    @Bean
    public Queue orderQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", "dlx_exchange");
        args.put("x-dead-letter-routing-key", "dlx_routing");
        args.put("x-message-ttl", 300000);  // TTL：5 分钟
        args.put("x-max-length", 10000);    // 队列最大长度
        return new Queue("order_queue", true, false, false, args);
    }
}