⭐⭐⭐ Spring Boot 项目实战	⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》	《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》	《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》	《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》	《Java 面试题 + Java 学习指南》

摘要: 原创出处君哥聊技术「朱晋君」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表

RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址

您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。

新的源码解析文章实时收到通知。每周更新一篇左右。

认真的源码交流微信群。

在使用消息队列时，有两个经常让我们烦恼的问题，消息丢失和消息重复。那我们在做技术选型时，有没有一个消息队列能解决消息丢失和消息重复这两个问题呢？

消息丢失

如上图，从生产者发送消息，Broker 保存消息，消费者消费消息，每一个环节都有可能丢失消息。

发送丢失

生产者发送消息时，如果处理不当，很可能会造成消息丢失。

生产者发送消息，主流消息队列都支持同步发送和异步发送。如果使用同步发送，生产者发送消息后，会同步等待 Broker 返回的 ACK，收到 ACK 消息，就认为消息发送成功。如果长时间没有收到，则会认为消息发送失败，需要进行重试。

同步发送可以保证消息不丢失，但是会有性能问题，所以多数情况会选择异步发送。异步发送如何保证消息不丢失呢？主流消息队列（比如 Kafka 和 RocketMQ）实现方法基本类似，使用回调函数来实现。下面看一下 Kafka 的异步发送代码：

producer.send(record, new Callback() {

public void onCompletion(RecordMetadata metadata, Exception exception) {
    if (exception != null) {
        logger.error("发送消息失败：", exception);
    }
    if (metadata != null) {
        logger.info("消息发送成功");
    }
    }
});

消息存储

生产者发送消息成功，也不能保证消息绝对不丢失。因为即使消息发送到 Broker，如果在消费者拉取到消息之前，Broker 宕机了，消息还没有落盘，也会导致消息丢失。

在存储阶段要保证消息不丢失，可以考虑几个方面：

同步刷盘

采用异步刷盘，如果在消息落盘之前 Broker 宕机了，就会造成消息丢失。而采用同步刷盘，等待消息落盘之后，再给 Sender 返回发送成功，可以从消息发送环节保证消息不丢失。

在 RocketMQ 中，把 flushDiskType 参数配置为 SYNC_FLUSH 就可以开启同步刷盘。

Broker 集群

如果 Broker 集群中只有一个节点，即使消息落盘成功了，Broker 发送故障，在 Broker 恢复以前消费者也会拉取不到消息。而且如果 Broker 磁盘故障不可恢复，消息也会丢失。

采用 Broker 集群可以很好地解决这个问题。见下图：

在 Broker 集群时，可以等待 2 个以上的节点同步消息完成后再给 Producer 返回成功。这样即使一个 Broker 挂了，也可以很容易找到替代的 Broker。

消息消费

消费者保证不丢失消息，需要消费完成后再给 Broker 返回 ACK。在主流的消息队列中，如果 Broker 收不到 ACK，都会给消费者再次发送这条消息。

有时候为了解决消息积压的问题，消费者拉取到消息后会直接返回 ACK，然后再异步执行消息处理逻辑。这样要保证消息不丢失，需要在返回 ACK 之前把消息保存到本地，比如持久化到数据库，后面可以取数据库保存的消息进行处理。

消息重复

消息重复一般有两个原因，一个是生产者发送消息后没有收到 ACK，然后进行重复发送，另一个原因是消费者消费完成后 Broker 没有收到 ACK，导致消息重复推送给消费者。

重复消息会对业务造成影响，比如电商场景中的重复支付、账务场景中的重复记账，对业务造成的影响都比较严重。

从目前主流的消息队列来看，并没有一个消息队列能解决消息重复消费的问题，只能在消费端做幂等处理。下面提供几个思路作为参考。

数据库唯一键约束

如果消息会落本地数据库，可以采用消息 ID 作为唯一键。如果消息不落数据库，可以将消息 ID 或者消息中其他唯一能标识消息的属性作为唯一键落业务数据表。

保存消费记录

我们也可以将消息 ID 保存 Redis，消费消息前判断消息 ID 是否已存在。

ValueOperations<String, String> valueOperations = redisTemplate.opsForValue();
Boolean result = valueOperations.setIfAbsent(messageId, messageId);
if (result) {
    //消费逻辑;
} else {
    logger.error("这条消息已经消费，跳过，消息ID:{}", messageId);
}

这里有一个注意点，如果消费失败了，需要删除 Redis 中保存的消息 ID。

总结

消息不丢失、不重复是消息队列的基本要求，但这个基本要求还是很难满足的。

消息丢失这个要求，主流消息队列通过消息重试和消息持久化的方式可以满足。

但消息重试也同时带来了消息重复的可能性，主流消息队列在解决重复消息的问题上并没有现成的方案，对不允许重复消费的场景，需要开发人员在消费端做幂等处理。

芋道源码 —— 纯源码解析博客

面试官：要保证消息不丢失，又不重复，消息队列怎么选型？