Kafka Producer 的 acks 参数：

acks=0   // 不等确认 → 性能最高，可能丢消息
acks=1   // Leader 写入就确认（默认）→ Leader 挂了可能丢
acks=all // 所有 ISR 副本都写入才确认 → 最安全

配合 min.insync.replicas=2：
// ISR 中至少 2 个副本存活才接受写入
// 如果只剩 1 个副本 → 拒绝写入（保护数据一致性）

// 生产者：指定 MessageQueueSelector
producer.send(msg, (mqs, msg, arg) -> {
    int index = Math.abs(orderId.hashCode()) % mqs.size();
    return mqs.get(index);
}, orderId);

// 消费者：MessageListenerOrderly（非 MessageListenerConcurrently）
consumer.registerMessageListener(new MessageListenerOrderly() {
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ...) {
        // 顺序消费逻辑
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

TransactionMQProducer producer = new TransactionMQProducer("tx-group");

// 设置事务监听器
producer.setTransactionListener(new TransactionListener() {

    // 执行本地事务（半消息发送成功后回调）
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            orderService.createOrder((OrderDTO) arg); // 本地事务
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    // 回查接口（Broker 主动查询事务状态）
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 查数据库确认本地事务是否执行成功
        boolean success = orderService.checkOrderExists(msg.getKeys());
        return success ? COMMIT_MESSAGE : ROLLBACK_MESSAGE;
    }
});

-- 去重表
CREATE TABLE msg_dedup (
  msg_id  VARCHAR(64) PRIMARY KEY,
  created_at DATETIME
);

-- 消费逻辑（伪代码）
@Transactional
public void consume(Message msg) {
    try {
        // 1. 去重：INSERT IGNORE（唯一键冲突会抛异常）
        dedupDao.insert(msg.getId());
    } catch (DuplicateKeyException e) {
        log.info("重复消息，跳过: {}", msg.getId());
        return; // 幂等：直接返回成功
    }

    // 2. 业务逻辑 + 状态机防重
    int rows = orderDao.updateStatus(
        orderId, "PAID", "UNPAID"  // WHERE status='UNPAID'
    );
    if (rows == 0) {
        log.info("订单状态已变更，跳过");
    }
}