@Service
public class ReliableProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendReliably(OrderMessage msg) {
        // syncSend 是同步阻塞的——返回 SEND_OK 才说明 Broker 已接收
        SendResult result = rocketMQTemplate.syncSend(
            "order-topic:created", msg);

        if (SendStatus.SEND_OK.equals(result.getSendStatus())) {
            log.info("消息已确认到达 Broker: msgId={}", result.getMsgId());
        } else {
            // 只有 SEND_OK 才认为成功——FLUSH_DISK_TIMEOUT 等状态说明刷盘超时
            log.error("消息发送未确认: status={}", result.getSendStatus());
            // 补偿：写入 DB 重试表
            saveToRetryTable(msg);
        }
    }
}

# broker.conf——同步刷盘
flushDiskType = SYNC_FLUSH

# broker.conf——同步主从
brokerRole = SYNC_MASTER

@Component
@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "order-consumer-group"
)
public class OrderRetryListener
        implements RocketMQListener<OrderMessage> {

    @Override
    public void onMessage(OrderMessage msg) {
        try {
            processOrder(msg);
            // 正常返回 → 消费成功 → offset 推进
        } catch (RetryableException e) {
            // 可重试异常 → 抛出去 → RocketMQ 自动重试
            throw new RuntimeException("临时失败，重试", e);
        } catch (NonRetryableException e) {
            // 不可重试异常（数据格式错误等）→ 记录日志并吞掉
            log.error("消息格式错误，跳过: orderId={}", msg.getOrderId(), e);
            // 正常返回 → 消费成功 → 这条坏消息被消费掉（不再重试）
        }
    }
}

@Component
@RocketMQMessageListener(
    topic = "%DLQ%order-consumer-group",   // ← 死信 Topic
    consumerGroup = "order-dlq-consumer-group"
)
public class OrderDLQListener
        implements RocketMQListener<MessageExt> {

    @Override
    public void onMessage(MessageExt msg) {
        // MessageExt 包含原始消息的全部信息
        log.error("死信消息: msgId={}, topic={}, tag={}, body={}, reconsumeTimes={}",
                msg.getMsgId(),
                msg.getTopic(),      // 原始 Topic
                msg.getTags(),       // 原始 Tag
                new String(msg.getBody()),
                msg.getReconsumeTimes()  // 重试次数
        );
        // 发告警、记录 DB、通知人工处理...
    }
}

@Component
@RocketMQMessageListener(
    topic = "order-topic",
    consumerGroup = "order-consumer-group"
)
public class IdempotentOrderListener
        implements RocketMQListener<MessageExt> {  // 注意：泛型是 MessageExt 以获取 msgId

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Override
    public void onMessage(MessageExt msg) {
        // RocketMQ 每条消息有全局唯一的 msgId
        String msgId = msg.getMsgId();
        String orderId = msg.getKeys();    // 发送时 setKeys 设置的业务 Key
        String idempotentKey = orderId != null ? orderId : msgId;

        // SETNX 原子判重
        Boolean firstTime = redisTemplate.opsForValue()
                .setIfAbsent("rocketmq:consumed:" + idempotentKey,
                        "1", Duration.ofHours(24));

        if (Boolean.FALSE.equals(firstTime)) {
            log.warn("重复消息，跳过: msgId={}, keys={}", msgId, orderId);
            return;  // 正常返回 → offset 推进
        }

        try {
            OrderMessage orderMsg = JSON.parseObject(
                    new String(msg.getBody()), OrderMessage.class);
            processOrder(orderMsg);
        } catch (Exception e) {
            // 处理失败 → 删除幂等标记，让重试时可以重新处理
            redisTemplate.delete("rocketmq:consumed:" + idempotentKey);
            throw new RuntimeException("消费失败，回滚幂等标记", e);
        }
    }
}

// 发送时设置业务 Key
Message<String> message = MessageBuilder
        .withPayload(JSON.toJSONString(orderMsg))
        .setHeader(MessageConst.PROPERTY_KEYS, "order:10001:created")
        .build();
rocketMQTemplate.syncSend("order-topic:created", message);