// ErrorMessageConfig（死信交换机与队列）
@Bean
public TopicExchange errorMessageExchange() {
    return new TopicExchange(Constants.MQ.Exchanges.ERROR, true, false);  // durable
}
@Bean
public Queue errorQueue() {
    //【前缀+微服务】作为队列名，开启持久化
    return new Queue(Constants.MQ.Queues.ERROR_PREFIX + appName, true);
}

var message = MessageBuilder.withBody(StrUtil.bytes(jsonMsg, CharsetUtil.CHARSET_UTF_8))
        .setDeliveryMode(MessageDeliveryMode.PERSISTENT)    // 持久化
        .build();

// MQServiceImpl#sendMsg
var msgId = IdUtil.simpleUUID();
var correlationData = new CorrelationData(msgId);         // 关联数据

// 添加 callback
correlationData.getFuture().addCallback(
        result -> {
            if (result.isAck()) {
                log.debug("消息发送成功, ID:{}", correlationData.getId());
            } else {
                // 3.2 nack，消息失败 → 落库等待补偿
                log.error("消息发送失败, ID:{}, 原因:{}", correlationData.getId(), result.getReason());
                var failMsgEntity = FailMsgEntity.builder()
                        .exchange(exchange)
                        .msgId(correlationData.getId())
                        .msg(jsonMsg)
                        .reason(result.getReason()).build();
                this.failMsgService.save(failMsgEntity);
            }
        },
        ex -> {
            // 发送后无法连接到 broker 等未知异常
            var exceptionInfo = ExceptionUtil.getMessage(ex);
            var failMsgEntity = FailMsgEntity.builder()
                    .exchange(exchange)
                    .msgId(correlationData.getId())
                    .msg(jsonMsg)
                    .reason(exceptionInfo).build();
            this.failMsgService.save(failMsgEntity);
        }
);

@Retryable(value = MsgException.class, maxAttempts = 3,
           backoff = @Backoff(value = 3000, multiplier = 1.5))
@Override
public Boolean sendMsg(String exchange, String routingKey, Object msg, int delay) {
    ...
    try {
        this.rabbitTemplate.convertAndSend(exchange, routingKey, message, mpp -> {
            mpp.getMessageProperties().setDelay(delay);
            return mpp;
        }, correlationData);
    } catch (Exception e) {
        // 转化为自定义异常 MsgException，主要是为了触发重试机制
        throw MsgException.builder()
                .failMsgEntity(FailMsgEntity.builder()...)
                .build();
    }
    return true;
}

/**
 * 3次失败后 存入数据库
 */
@Recover
public Boolean saveFailMag(MsgException msgException) {
    this.failMsgService.save(msgException.getFailMsgEntity());
    return true;
}

// MsgException
@Data @Builder
public class MsgException extends RuntimeException {
    private FailMsgEntity failMsgEntity;   // 直接把待落库的数据挂在异常上
}

CREATE TABLE `cc_fail_msg` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `msg_id`      varchar(32)  DEFAULT NULL COMMENT '消息ID',
  `exchange`    varchar(100) DEFAULT NULL COMMENT '交换机',
  `routing_key` varchar(50)  DEFAULT NULL COMMENT '路由键',
  `msg`         text         COMMENT '消息内容',
  `reason`      text         COMMENT '失败原因',
  `created`     datetime     DEFAULT CURRENT_TIMESTAMP,
  `updated`     datetime     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `idx_created` (`created`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

// FailMsgJob
@Slf4j
@Component
@ConditionalOnBean({MQService.class, FailMsgService.class})
public class FailMsgJob {

    @XxlJob("failMsgJob")
    public void execute() {
        // 查询失败的数据，每次最多处理100条
        var queryWrapper = new LambdaQueryWrapper<FailMsgEntity>()
                .orderByAsc(FailMsgEntity::getCreated)     // 先进先出
                .last("limit 100");
        var failMsgEntityList = this.failMsgService.list(queryWrapper);
        if (CollUtil.isEmpty(failMsgEntityList)) {
            return;
        }
        for (var failMsgEntity : failMsgEntityList) {
            try {
                this.mqService.sendMsg(failMsgEntity.getExchange(),
                        failMsgEntity.getRoutingKey(), failMsgEntity.getMsg());
                this.failMsgService.removeById(failMsgEntity.getId());   // 发成功就删
            } catch (Exception e) {
                log.error("处理错误消息失败, failMsgEntity = {}", failMsgEntity, e);
            }
        }
    }
}

// ErrorMessageConfig
@Value("${spring.application.name}")
private String appName;

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate) {
    return new RepublishMessageRecoverer(rabbitTemplate, Constants.MQ.Exchanges.ERROR, appName);
}

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto        # 异常时 nack，消息回滚
        retry:
          enabled: true               # 开启消费者失败重试
          initial-interval: 1000      # 初始等待 1 秒
          multiplier: 1
          max-attempts: 3             # 最大重试 3 次
          stateless: true             # 无状态重试

消费失败
  → 间隔 1s 重试（第 1 次）
  → 间隔 1s 重试（第 2 次）
  → 间隔 1s 重试（第 3 次）
  → MessageRecoverer 触发
  → 重新发到 cc.exchange.topic.error，routingKey = ${spring.application.name}
  → 路由到 cc.queue.error.chain-cloud-ms-oms
  → 等待人工处理 / 告警

} catch (Exception e) {
    throw MsgException.builder().failMsgEntity(...).build();
}

// 可以重试：网络类
if (e instanceof AmqpConnectException || e instanceof AmqpIOException) {
    throw new MsgRetryableException(...);
}
// 不可重试：配置/路由类，直接落库不重试
if (e instanceof AmqpIllegalStateException) {
    this.failMsgService.save(...);     // 直接落库
    return true;
}

convertAndSend 成功
  → broker 收到，exchange 存在
  → 但 routingKey 没有匹配的 binding
  → 消息被 broker 丢弃
  → confirm 仍然返回 ack（因为 exchange 确实收到了）

// MessageConfig
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    var rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);
    rabbitTemplate.setReturnsCallback(message -> {
        // 延迟消息没有发到队列是正常情况，无需记录日志
        if (StrUtil.contains(message.getExchange(), Constants.MQ.DELAYED_KEYWORD)) {
            return;
        }
        // 投递失败，记录日志
        log.error("消息没有投递到队列，应答码：{}，原因：{}，交换机：{}，路由键：{},消息：{}",
                message.getReplyCode(), message.getReplyText(),
                message.getExchange(), message.getRoutingKey(), message.getMessage());
    });
}

spring:
  rabbitmq:
    publisher-returns: true        # 开启 publisher-return
    template:
      mandatory: true              # 必须显式声明！否则不路由的消息会被静默丢弃

if (StrUtil.contains(message.getExchange(), Constants.MQ.DELAYED_KEYWORD)) {
    return;    // 延迟消息没有发到队列是正常情况
}

// TransportOrderDispatchMQListener 的 Javadoc
 * 核心职责：
 * 1. 消费MQ中的运单消息
 * 2. 做幂等控制（防止重复消费）      ← 写了，没做
 * 3. 调用AI服务计算优先级
 * 4. 将运单写入Redis ZSet，实现优先级排序

// OMS：只是过滤掉没有状态的消息，不是幂等
if (ObjectUtil.isEmpty(transportOrderStatusMsg.getStatusCode())) { return; }

// OMS 支付：过滤出"已结算"的，重复消息会重复 update（虽然 update 是幂等写）
var msgList = tradeStatusMsgList.stream()
        .filter(v -> v.getStatusCode().equals(TradingStateEnum.YJS.getCode()))
        .collect(Collectors.toList());

// track：先查后插，并发下不安全
if (ObjectUtil.isNotEmpty(trackEntity)) {
    throw new CCException(TrackExceptionEnum.TRACK_ALREADY_EXISTS);
}

// 第 1 层：业务状态机天然幂等（最优）
// 状态流转只允许"从 A 到 B"，重复到达 B 时不产生新动作
public boolean updateStatus(Long id, DriverJobStatusEnum target) {
    LambdaUpdateWrapper<DriverJobEntity> w = Wrappers.lambdaUpdate();
    w.eq(DriverJobEntity::getId, id)
     .eq(DriverJobEntity::getStatus, DriverJobStatusEnum.PROCESSING.getCode())  // ← 乐观锁式的条件更新
     .set(DriverJobEntity::getStatus, target.getCode());
    return super.update(w);   // 返回 false 说明状态已经变了，直接忽略
}

// 第 2 层：唯一索引兜底
// 消费时如果插入遇到唯一键冲突，识别为"已处理过"，直接 ack
try {
    this.taskService.save(task);
} catch (DuplicateKeyException e) {
    log.info("任务已存在，跳过：{}", task.getId());
}

// 第 3 层：消息去重表（跨服务通用）
// INSERT INTO cc_msg_consume(msg_id, consumer) VALUES (?, ?)
//   → DuplicateKeyException = 重复消费 → 直接返回

// MQServiceImpl 三处 FailMsgEntity.builder()
// 第 1 处：nack 回调（约 76-80 行）
var failMsgEntity = FailMsgEntity.builder()
        .exchange(exchange)
        .msgId(correlationData.getId())
        .msg(jsonMsg)
        .reason(result.getReason()).build();
//        ↑ 没有 .routingKey(...)

// 第 2 处：发送异常回调（约 88-92 行）
var failMsgEntity = FailMsgEntity.builder()
        .exchange(exchange)
        .msgId(correlationData.getId())
        .msg(jsonMsg)
        .reason(exceptionInfo).build();
//        ↑ 也没有

// 第 3 处：MsgException 里（约 111-115 行）
.failMsgEntity(FailMsgEntity.builder()
        .exchange(exchange)
        .msgId(correlationData.getId())
        .msg(jsonMsg)
        .reason(ExceptionUtil.getMessage(e)).build())
//        ↑ 同样没有

// 实体定义
@TableName("cc_fail_msg")
public class FailMsgEntity extends BaseEntity {
    private String msgId;
    private String exchange;
    private String routingKey;      // ← 字段是有的，就是不赋值
    private String msg;
    private String reason;
}

发送失败 → 落库 cc_fail_msg，routing_key = NULL
  → failMsgJob 读出来
  → sendMsg(exchange, null, msg)
  → TOPIC 交换机用 null 路由 → 匹配不到任何 binding
  → 消息再次被丢弃
  → 但 sendMsg 返回 true（乐观返回）
  → failMsgJob 执行 removeById，**把这条记录删掉了**
  → 消息永久丢失，且没有任何痕迹

var failMsgEntity = FailMsgEntity.builder()
        .exchange(exchange)
        .routingKey(routingKey)          // ← 加这一行
        .msgId(correlationData.getId())
        .msg(jsonMsg)
        .reason(result.getReason()).build();

@Retryable(value = MsgException.class, maxAttempts = 3, backoff = @Backoff(value = 3000, multiplier = 1.5))
@Override
public Boolean sendMsg(String exchange, String routingKey, Object msg) {
    // 实时发送
    return this.sendMsg(exchange, routingKey, msg, Constants.MQ.DEFAULT_DELAY);   // ← this. 自调用
}

@Retryable(value = MsgException.class, maxAttempts = 3, backoff = @Backoff(value = 3000, multiplier = 1.5))
@Override
public Boolean sendMsg(String exchange, String routingKey, Object msg, int delay) {
    ...
}

@Autowired @Lazy private MQService self;   // 注入代理而不是自己

@Override
public Boolean sendMsg(String exchange, String routingKey, Object msg) {
    return self.sendMsg(exchange, routingKey, msg, Constants.MQ.DEFAULT_DELAY);
}

try {
    this.rabbitTemplate.convertAndSend(exchange, routingKey, message, ...);
} catch (Exception e) {
    throw MsgException.builder()...;
}
return true;      // ← 没有等 confirm 就返回 true

this.mqService.sendMsg(...);              // 乐观返回 true
this.failMsgService.removeById(...);      // 删掉失败记录

// 方案 1：补偿任务里等 confirm 结果（需要改 API 语义）
// 方案 2：补偿任务不删记录，改为标记状态，靠下一个轮次重试
//         发送成功后由 confirm 回调把记录标记为 SUCCESS
// 方案 3（最简单、最可靠）：给 cc_fail_msg 加 retry_count，
//   超过阈值就不再重试但保留记录，同时告警

oms-service 发送消息 → 通过 MQFeign（Feign 调 base-service 的 /mq 接口）
                                    ↓
                        base-service 执行发送 + 重试 + 落库 cc_fail_msg
                                    ↓
                        base-service 的 failMsgJob 做补偿

chain-cloud-ms-mq  ——  独立的"消息网关"服务
  ├── 提供 HTTP/SDK 接口
  ├── 自己持有 cc_fail_msg 表
  ├── 自己注册 XXL-Job 执行器
  └── 其它服务通过它发消息，但账号/队列按发送方隔离

correlationData.getFuture().addCallback(result -> {
    ...
    this.failMsgService.save(failMsgEntity);    // 在 Rabbit 客户端线程上执行
}, ...);

// 把 confirm 回调交给独立的线程池，不要占用 Rabbit IO 线程
@Bean("mqConfirmExecutor")
public Executor mqConfirmExecutor() {
    ThreadPoolTaskExecutor e = new ThreadPoolTaskExecutor();
    e.setCorePoolSize(2);
    e.setMaxPoolSize(8);
    e.setQueueCapacity(1000);
    e.setThreadNamePrefix("mq-confirm-");
    e.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    return e;
}

@Resource(name = "mqConfirmExecutor")
private Executor mqConfirmExecutor;
...
correlationData.getFuture().addCallback(result -> {
    mqConfirmExecutor.execute(() -> { /* 落库 */ });
}, ...);

当前方案（"发消息 + 失败补偿"）：
  BEGIN
    INSERT cc_order           ← 事务 1
  COMMIT
  sendMsg(...)                ← 事务外
       ↓ 失败 → cc_fail_msg → XXL-Job 重发
  问题：① 两个事务之间有窗口  ② 补偿可能无限重试  ③ 需要额外表

Outbox 方案（"消息也进事务"）：
  BEGIN
    INSERT cc_order
    INSERT cc_outbox(msg, exchange, routingKey, status=PENDING)   ← 同一个事务！
  COMMIT
  ↓
  轮询任务（或 CDC/binlog 订阅）：
     SELECT * FROM cc_outbox WHERE status=PENDING LIMIT 100
     → 发消息 → 等 confirm
     → ack: UPDATE status=SUCCESS
     → nack: UPDATE retry_count++, 超阈值则 status=DEAD

1. 交换机/队列持久化 + 消息 PERSISTENT
2. Publisher Confirm（CorrelationData 异步回调）
3. Publisher Returns（mandatory + returnsCallback，捕捉"路由不到队列"）
4. 发送异常 Spring Retry（3 次，退避 3s×1.5）+ @Recover 落库 cc_fail_msg
5. XXL-Job failMsgJob 定时补偿（100 条/次，先进先出）
消费侧：3 次重试 + RepublishMessageRecoverer 进 cc.queue.error.<服务名>

1. DB 提交和消息发送之间有窗口（我们用"延迟 5 秒"来缓解，但这只是缓解）
2. 补偿任务可能无限重试（没有 retry_count）
3. 没有"待发消息"的可观测视图（出问题时只能看日志）

第 1 层（最优）：业务状态机幂等
   状态流转用"条件更新"—— UPDATE ... WHERE id=? AND status=?
   重复消息到达时 WHERE 不匹配，affected rows = 0，直接忽略
   这层是免费的，而且符合业务语义

第 2 层：唯一索引兜底
   依赖数据库保证。比如运单唯一、任务 (order_id, task_type) 唯一
   插入时捕获 DuplicateKeyException，识别为"已处理"直接 ack

第 3 层：消息去重表
   INSERT INTO cc_msg_consume(msg_id, consumer) —— 唯一键冲突即重复
   通用中间件方案，但多一次 IO

messagePostProcessor.getMessageProperties().setDelay(delay);

@Exchange(name = "...", type = ExchangeTypes.TOPIC, delayed = Constants.MQ.DELAYED)