// 同步发送（等待Broker ACK）
SendResult result = producer.send(message);
// SendStatus.SEND_OK: 发送成功
// SendStatus.FLUSH_DISK_TIMEOUT: 刷盘超时（消息可能丢失）
// SendStatus.SLAVE_NOT_AVAILABLE: 从节点不可用
if (result.getSendStatus() != SendStatus.SEND_OK) {
    // 告警，记录到数据库，后续补偿
}

Broker端：同步刷盘

# RocketMQ配置：同步刷盘（默认是异步）
flushDiskType=SYNC_FLUSH

同步刷盘：消息写入后，等待落盘完成才返回ACK，100%不丢，但写入延迟增加。异步刷盘：消息写入PageCache就返回ACK，定期批量落盘，性能好但有丢失风险。

高可用场景还需要主从复制：消息同步到从节点后才算发送成功（同步复制）。

Consumer端：手动ACK

消费者处理完消息后才发送ACK，如果处理中失败（未ACK），Broker会重新投递。

// RocketMQ消费者：手动ACK
@RocketMQMessageListener(topic = "order_topic", consumerGroup = "order_group")
public class OrderConsumer implements RocketMQListener<String> {
    
    @Override
    public void onMessage(String message) {
        try {
            processOrder(message); // 业务处理
            // 方法正常返回 = ACK确认
        } catch (Exception e) {
            log.error("消息处理失败", e);
            // 抛出异常 = NACK，Broker稍后重新投递
            throw e;
        }
    }
}

2.3 保证消息不重复：幂等消费

消息队列通常保证At-least-once（至少一次），而不是Exactly-once（精确一次）。重复投递不可避免，消费者必须实现幂等。

幂等实现方案：

@Service
public class IdempotentConsumer {
    
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    @Autowired
    private OrderService orderService;
    
    public void consume(Message msg) {
        String msgId = msg.getMsgId(); // RocketMQ的消息ID
        String key = "consumed_msg:" + msgId;
        
        // SET key 1 NX EX 86400：不存在才设置（原子操作），24小时过期
        Boolean first = redisTemplate.opsForValue().setIfAbsent(key, "1", 
            24, TimeUnit.HOURS);
        
        if (!Boolean.TRUE.equals(first)) {
            log.info("消息{}已消费，幂等跳过", msgId);
            return; // 幂等处理
        }
        
        try {
            orderService.processOrder(msg.getBody());
        } catch (Exception e) {
            // 处理失败，删除幂等key（允许重试）
            redisTemplate.delete(key);
            throw e;
        }
    }
}

2.4 顺序消费

为什么顺序消费难？

Broker通常有多个Queue（RocketMQ）或Partition（Kafka），消息可能被分散到不同的Queue/Partition，由不同的消费者并行处理，顺序无法保证。

局部顺序（推荐）：

让同一个订单的所有消息，路由到同一个Queue，由同一个消费者线程顺序处理。

// Producer：同一订单ID的消息发到同一个Queue
producer.send(message, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object orderId) {
        // 根据orderId取模，映射到固定的Queue
        int index = Math.abs(orderId.hashCode()) % mqs.size();
        return mqs.get(index);
    }
}, orderId); // orderId作为selector的参数

// Consumer：顺序消费监听器（RocketMQ）
@Service
public class OrderedConsumer implements MessageListenerOrderly {
    
    @Override
    public ConsumeOrderlyStatus consumeMessage(
            List<MessageExt> msgs, ConsumeOrderlyContext context) {
        for (MessageExt msg : msgs) {
            // MessageListenerOrderly确保同一Queue中的消息顺序消费
            System.out.println("顺序消费: " + new String(msg.getBody()));
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
}

三、标准答案 + 代码验证

3.1 RocketMQ完整可靠发送

@Service
public class ReliableMessageProducer {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    @Autowired
    private MessageLogMapper messageLogMapper; // 消息日志表
    
    /**
     * 可靠发送：先保存消息日志，再发送，失败则由定时任务补偿
     */
    @Transactional
    public void sendReliable(String topic, Object payload, String businessId) {
        // 1. 先把消息保存到数据库（状态：待发送）
        // 这步在事务中，保证业务操作和消息记录的原子性
        MessageLog log = new MessageLog();
        log.setTopic(topic);
        log.setPayload(JSON.toJSONString(payload));
        log.setBusinessId(businessId);
        log.setStatus(0); // 0=待发送
        log.setCreatedAt(new Date());
        messageLogMapper.insert(log);
        
        // 2. 发送消息（在事务提交后执行，用TransactionSynchronization）
        TransactionSynchronizationManager.registerSynchronization(
            new TransactionSynchronizationAdapter() {
                @Override
                public void afterCommit() {
                    try {
                        rocketMQTemplate.syncSend(topic, payload);
                        // 发送成功，更新状态
                        messageLogMapper.updateStatus(log.getId(), 1); // 1=已发送
                    } catch (Exception e) {
                        log.error("消息发送失败，等待定时任务补偿", e);
                        // 不抛出异常，不影响业务事务
                        // 定时任务会定期扫描status=0且超时的消息重发
                    }
                }
            }
        );
    }
}

// 定时任务补偿：扫描发送失败的消息
@Component
public class MessageRetryTask {
    
    @Scheduled(fixedDelay = 60000) // 每分钟执行
    public void retry() {
        // 查询5分钟前还未发送的消息
        Date threshold = new Date(System.currentTimeMillis() - 5 * 60 * 1000);
        List<MessageLog> pendingLogs = messageLogMapper.findPending(threshold, 100);
        
        for (MessageLog log : pendingLogs) {
            try {
                rocketMQTemplate.syncSend(log.getTopic(), log.getPayload());
                messageLogMapper.updateStatus(log.getId(), 1);
            } catch (Exception e) {
                log.error("消息重发失败: {}", log.getId(), e);
            }
        }
    }
}

3.2 Kafka的顺序消费

// Kafka Producer：同一key的消息发到同一Partition
@Service
public class KafkaOrderProducer {
    
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    
    public void sendOrder(String orderId, String event) {
        // key = orderId，相同orderId的消息路由到同一Partition
        // Kafka保证同一Partition内消息有序
        kafkaTemplate.send("order_topic", orderId, event);
    }
}

// Kafka Consumer：单线程处理每个Partition
@KafkaListener(topics = "order_topic", groupId = "order_group",
    concurrency = "3") // 3个并发，对应3个Partition
public void consume(ConsumerRecord<String, String> record) {
    // 同一Partition的消息由同一线程顺序消费
    log.info("Partition: {}, Offset: {}, Key: {}, Value: {}",
        record.partition(), record.offset(), record.key(), record.value());
    processOrder(record.value());
}

我的回答：RocketMQ专门为金融级可靠性设计，支持同步刷盘、主从同步复制，消息不丢失的保证更强。同时支持事务消息（半消息机制），保证生产者本地事务和消息发送的原子性。Kafka默认是异步刷盘，消息可靠性稍弱，但吞吐量极高（百万/秒级别）。Kafka的acks=all配置要求所有ISR副本都确认才算发送成功，可以提高可靠性，但延迟增加。总体来说，对可靠性要求极高（支付、金融）用RocketMQ，对吞吐量要求极高（日志收集、大数据）用Kafka。

追问2：RocketMQ的事务消息是如何实现的？

我的回答：事务消息解决的是"本地事务和发消息的原子性"问题。过程分三步：第一步，发送半消息（Half Message）到Broker，Broker存储但不投递给消费者。第二步，执行本地事务（如数据库操作），执行成功则发送Commit给Broker，失败则发送Rollback。第三步，如果Broker长时间没收到Commit或Rollback（Producer异常），Broker会回调Producer的checkLocalTransaction方法，由业务方再次确认本地事务状态，决定是Commit还是Rollback。这样确保了本地事务和消息发送要么都成功，要么都失败。

追问3：消息队列积压了怎么处理？

我的回答：消息积压说明消费速度跟不上生产速度，处理思路有几种。第一，临时扩容消费者，增加消费者数量（最直接），但要注意数据库等下游系统的承受能力。第二，跳过非关键消息，如果积压消息中有历史的、不重要的消息（如旧的日志），可以设置消费位移跳过这些消息，只处理最新的。第三，提升单个消费者处理能力，使用批量消费（一次拉取多条消息批量处理），减少网络往返次数。第四，消费者处理后不立即ACK，批量ACK，减少频繁提交offset的开销。对于Kafka，还可以临时增加Partition数量，让更多消费者并行处理（注意Partition数量只能增不能减）。

五、同类题目举一反三

消息队列如何实现延迟消息（定时消息）？

RocketMQ原生支持18个延迟级别（1s/5s/10s/...），通过延迟队列实现，Producer发送时指定延迟级别，Broker在到期后才投递给消费者。Kafka不原生支持延迟消息，通常的方案是：单独的延迟Topic，消费者轮询到期消息后转发到正式Topic。RabbitMQ通过死信队列+TTL实现延迟：消息先发到TTL队列，超时后成为死信，被路由到正式队列。

六、踩坑实录

坑一：消费者自动提交offset，处理失败导致消息丢失

Kafka消费者配置了enable.auto.commit=true（默认），消息从Broker拉取后，定时自动提交offset。如果消息还在处理中，offset已经提交，这时消费者重启或异常，从新的offset开始消费，就会跳过未处理完的消息。修复：改为手动提交，处理完一批消息后手动consumer.commitSync()。

坑二：顺序消费中，某条消息处理失败导致整个Queue阻塞

顺序消费中，RocketMQ的MessageListenerOrderly在某条消息处理失败时，会不断重试这条消息（直到成功），导致后续消息全部阻塞。如果消息是因为业务逻辑Bug永远处理不成功，整个Queue就永久阻塞了。解决方案：设置最大重试次数，超过后把消息转移到死信队列（DLQ），人工处理。

坑三：RocketMQ消费组名写错，消费者收不到消息

新服务上线，消费者配置的consumerGroup和Producer发送的topic匹配了，但消费者订阅的topic名字多了个空格，一直收不到消息。消息队列的配置（topic、tag、consumerGroup）要严格对齐，建议用常量类统一管理，避免硬编码字符串导致的typo问题。

七、总结

消息队列三大保证的核心方案：

不丢失：

Producer：同步发送 + 失败重试 + 消息日志表兜底
Broker：同步刷盘 + 主从同步复制
Consumer：手动ACK + 处理失败重新投递

不重复：

消费者实现幂等（消息ID去重，Redis SET NX + 业务唯一键）
接受At-least-once，在消费端保证幂等

顺序消费：

局部有序（实际场景够用）：相同业务ID的消息路由到同一Queue/Partition
消费者顺序处理Queue/Partition内的消息（单线程或MessageListenerOrderly）
全局有序（性能代价大）：整个Topic只有一个Queue，性能极低