消息队列可靠性保障方案

// 方式一：发完即忘（fire-and-forget）—— 最高效，最不可靠
kafkaTemplate.send(topic, message);
// 不关心是否到达 broker，网络抖动就丢了

// 方式二：同步发送 —— 可靠，性能差
try {
    RecordMetadata metadata = kafkaTemplate.send(topic, message).get(3, TimeUnit.SECONDS);
    System.out.println("发送成功，offset=" + metadata.getOffset());
} catch (ExecutionException e) {
    // 发送失败，broker 没收到
    handleFailure(e);
}

// 方式三：异步发送 + 回调 —— 推荐
kafkaTemplate.send(topic, message, new Callback() {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception e) {
        if (e != null) {
            log.error("发送失败", e);
            // 记录失败，重试
            retrySend(topic, message);
        } else {
            log.info("发送成功 offset={}", metadata.getOffset());
        }
    }
});

@KafkaListener(topics = "payment-success", groupId = "order-service")
public void handlePaymentSuccess(ConsumerRecord<String, PaymentMessage> record, Acknowledgment ack) {
    PaymentMessage message = record.value();

    try {
        // 1. 业务处理
        orderService.confirmOrder(message.getOrderId());

        // 2. 业务处理成功后才提交 offset
        ack.acknowledge();
        log.info("消费成功 offset={}", record.offset());

    } catch (Exception e) {
        // 3. 业务处理失败，不提交 offset，消息会被重新消费
        log.error("消费失败 offset={}", record.offset(), e);
        // 可以在这里记录失败日志，便于排查
        throw e; // 重新抛出，让 Kafka 自动重试
    }
}

// Step 1: 发送半消息（Half Message）—— 对消费者不可见
@Transactional
public void createOrder(OrderCreateRequest request) {
    // 1. 发送半消息，事务开始
    TransactionMQProducer producer = rocketMQTemplate.getProducer();
    Message message = MessageBuilder
        .withPayload(request)
        .setHeader("orderId", request.getOrderId())
        .build();

    // 发送半消息，并注册事务执行器
    TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(
        "order-topic",
        message,
        new OrderTransactionListener(orderService, request) // 事务回调
    );
}

// Step 2: 事务执行器（本地事务 + 回查）
@Service
@Slf4j
public class OrderTransactionListener implements TransactionListener {

    @Autowired
    private OrderMapper orderMapper;

    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        OrderCreateRequest request = (OrderCreateRequest) arg;
        try {
            // 本地事务：创建订单（和半消息在同一个事务中）
            orderMapper.createOrder(request);
            return LocalTransactionState.COMMIT_MESSAGE; // 提交，消息对消费者可见
        } catch (Exception e) {
            log.error("本地事务失败，回滚", e);
            return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚，消息被丢弃
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // RocketMQ 自动回查：事务超时未提交，调用此方法
        String orderId = msg.getUserProperties("orderId");
        Order order = orderMapper.selectByOrderId(orderId);

        if (order != null) {
            return LocalTransactionState.COMMIT_MESSAGE; // 订单存在，提交
        } else {
            return LocalTransactionState.ROLLBACK_MESSAGE; // 订单不存在，回滚
        }
    }
}

// 消息处理记录表
@TableName("message_dedup")
public class MessageDedup {
    private String messageId;      // 消息唯一 ID
    private Integer status;        // 0-处理中 1-成功 2-失败
    private LocalDateTime createTime;
}

@Service
public class OrderService {

    @Autowired
    private MessageDedupMapper dedupMapper;

    public void handlePaymentSuccess(PaymentMessage msg) {
        String messageId = msg.getMessageId();

        // 1. 尝试插入幂等记录
        int inserted = dedupMapper.insertIfNotExists(messageId);
        if (inserted == 0) {
            log.info("消息已处理过，跳过: {}", messageId);
            return;
        }

        // 2. 执行业务
        processOrder(msg);

        // 3. 更新状态
        dedupMapper.updateStatus(messageId, 1);
    }
}

public void handleMessage(String messageId) {
    String key = "msg:consumed:" + messageId;

    // SETNX 原子操作
    Boolean success = redisTemplate.opsForValue()
        .setIfAbsent(key, "1", Duration.ofHours(24));

    if (!Boolean.TRUE.equals(success)) {
        log.info("消息已消费，跳过: {}", messageId);
        return;
    }

    // 业务处理...
}

// 生产者：指定相同 key 的消息发送到同一分区
kafkaTemplate.send("order-topic", orderId, orderMessage);
// 同一 orderId 的消息一定到同一分区，分区内有序

// 消费者：单分区单线程消费
@KafkaListener(topics = "order-topic", containerFactory = "kafkaListenerContainerFactory")
public void consume(ConsumerRecord<String, OrderMessage> record) {
    // 同一分区的消息顺序消费
    processOrder(record.value());
}

// 死信队列名称自动生成：DLQ.topic-name
// 消费死信队列消息，处理异常情况
@RocketMQMessageListener(
    topic = "DLQ-order-topic",
    consumerGroup = "dlq-handler-group"
)
public class DlqHandler {
    public void handle(MessageExt msg) {
        log.error("死信消息: {}", new String(msg.getBody()));
        // 人工处理：记录日志、发告警、人工补偿
    }
}