RocketMQ和Kafka的区别，以及如何保证消息不丢失和重复消费

消息队列RocketMQ和Kafka的区别

性能（单台）语言多语言支持客户端优缺点RocketMQ十万级javajava

• 模型简单、接口易用，在阿里有大规模应用

• 文档少，支持的语言少
  Kafka百万级服务端scala，客户端java主流语言均支持

• 天生分布式、性能最好，常用于大数据领域

• 运维难度大，对zookeeper强依赖，多副本机制下对带宽有一定要求

• RocketQA适用于高性能、高可用的消息传递场景，具有丰富的消息过滤和分布式事务特性；

• Kafka适用于高吞吐量、低延迟的实时数据处理和事件驱动的架构场景，具有良好的可伸缩性和持久性

Kafka 消费顺序、消息丢失和重复消费

Kafka 采用的是发布 - 订阅模型。

RocketMQ 的消息模型和 Kafka 基本是完全一样的。唯一的区别是 Kafka 中没有队列这个概念，与之对应的是 Partition（分区）。

Kafka的多副本机制

Kafka 为分区（Partition）引入了多副本（Replica）机制。

分区（Partition）中的多个副本之间会有一个叫做 leader 的家伙，其他副本称为 follower。我们发送的消息会被发送到 leader 副本，然后 follower 副本才能从 leader 副本中拉取消息进行同步。生产者和消费者只与leader副本做交互。

Kafka 的多分区（Partition）以及多副本（Replica）机制有什么好处呢？

1. Kafka 通过给特定 Topic 指定多个 Partition, 而各个 Partition 可以分布在不同的 Broker 上, 这样便能提供比较好的并发能力（负载均衡）。
2. Partition 可以指定对应的 Replica 数, 这也极大地提高了消息存储的安全性, 提高了容灾能力，不过也相应的增加了所需要的存储空间。

Kafka如何保证消息的顺序消费

• 一个topic只对应一个partition
• （推荐）发送消息的时候指定key/partition【采用对象的id做为key】

Kafka 中发送 1 条消息的时候，可以指定 topic, partition, key,data（数据） 4 个参数。如果你发送消息的时候指定了 Partition 的话，所有消息都会被发送到指定的 Partition。并且，同一个 key 的消息可以保证只发送到同一个 partition，这个我们可以采用表/对象的 id 来作为 key 。

如何保证消息不丢失

• 生产者消息丢失

生产者消息是如何丢失的？

生产者调用send方法发送消息之后【异步操作】，消息可能因为网络原因没有发出去

为了确定消息发送是否成功，需要判断消息发送的结果。

一般采用如下做法，采用回调函数获取消息发送的状态,并且为生产者设置重试次数和重试间隔，一般为3

@Slf4j
public class KafkaDemo {
    @Autowired
    KafkaTemplate kafkaTemplate;

    public void sendMessage(String topic,Object o) throws ExecutionException, InterruptedException {
        ListenableFuture<SendResult<String, Object>> future= kafkaTemplate.send(topic,o);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("生产者发送消息：{} 失败，原因：{}", o.toString(), ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("生产者成功发送消息到topic:{} partition:{}的消息",result.getProducerRecord().value().toString());
            }
        });
    }
}

• 消费者消息丢失

消息在被追加到 Partition(分区)的时候都会分配一个特定的偏移量（offset）。偏移量（offset)表示 Consumer 当前消费到的 Partition(分区)的所在的位置。Kafka 通过偏移量（offset）可以保证消息在分区内的顺序性。

消费者消息是如何丢失的？

当消费者拉取到了分区的某个消息之后，消费者会自动提交了 offset。自动提交的话会有一个问题，试想一下，当消费者刚拿到这个消息准备进行真正消费的时候，突然挂掉了，消息实际上并没有被消费，但是 offset 却被自动提交了。

解决方案：

关闭自动提交offset，每次在真正消费完消息之后再手动提交offset。

但是这样也会带来重复消费的情况，比如，消费了一半还没提交offset突然挂掉，那么这个消息理论上会被消费两次，这种情况怎么办？【保证幂等性】

• kafka丢失了消息

kafka是如何丢失消息的？【多副本机制】

假如 leader 副本所在的 broker 突然挂掉，那么就要从 follower 副本重新选出一个 leader ，但是 leader 的数据还有一些没有被 follower 副本的同步的话，就会造成消息丢失。

1）acks=all 所有副本全部收到消息时，生产者才会接收到来自服务器的响应

1. replication.factor >= 3 每个分区至少有3个副本

2. min.insync.replicas > 1 消息至少被写入2个副本，才算是被成功发送

4）unclean.leader.election.enable = false 当leader副本发生故障时不会从followers副本中和leader副本同步程度达不到要求的副本中选出leader，降低了消息丢失的可能性。

如何保证消息不被重复消费

kafka出现重复消费的原因？

1）消费者已经消费了的消息没有正确提交offset（根本原因）

2）kafka侧因为服务端业务处理时间过长或者网络连接等原因，让kafka认为服务假死，触发了分区rebalance

解决方案

• 消费者做幂等校验【如redis的set、mysql的主键等天然的幂等功能】
• 关闭自动提交,enable.auto.commit=false，开发者在代码中手动提交offset - 消费完消息再提交：依旧有重复消费的风险，和自动提交一样- 拉取到消息即提交：会有消息丢失的风险，允许消息延时的场景，一般采用这种方式。通过定时任务在业务不繁忙的时候做数据兜底。

Kafka失败重试机制

kafka消费者默认配置下最多重试10次，每次时间间隔0，即立即重试。如果在 10 次重试后仍然无法成功消费消息，则不再进行重试，消息将被视为消费失败。

RocketMQ顺序消费、消息丢失和重复消费

生产者组中的生产者会向主题发送消息，而 主题中存在多个队列，生产者每次生产消息之后是指定主题中的某个队列发送消息的。

RocketMQ

通过**使用在一个

Topic

中配置多个队列并且每个队列维护每个消费者组的消费位置** 实现了 主题模式/发布订阅模式

顺序消费

RocketMQ在topic上是无序的，只有在队列层面才能保证是有序的。

【把同一主题消息放入相同的队列】

顺序需要由3个阶段去保障：

• 消息被发送时保持顺序 - 将同一订单的消息发送到相同的队列【通过MessageQueueSelector来实现队列的选择】，通过对订单的唯一标识做Hash，将同一个订单的消息发送到相同的队列
• 消息被存储时保持和发送时顺序一致
• 消息被消费时保持和存储时的顺序一致 - 同一个订单的消息被同一个消费者消费【订单id做hash选择相同的队列】- 消费者消费完一条消息之后，才能接着消费下一条【单线程消费】

消息不丢失、可靠性保证

• 发送端丢失消息 - 发送端在发送消息时，传入回调接口实现类，调用该发送接口后不会阻塞，发送方法会立即返回，回调任务会在另一个线程中执行，消息发送结果会回传给相应的回调函数。具体的业务实现可以根据发送的结果信息来判断是否需要重试来保证消息的可靠性。
• 存储端丢失消息 - 主从复制保证
• 消费端丢失消息 - 消费重试- 死信队列 - 未能成功消费的消息，消息队列并不会立刻将消息丢弃，而是将消息发送到死信队列，其名称是在原队列名称前加%DLQ%，如果消息最终进入了死信队列，则可以通过RocketMQ提供的相关接口从死信队列获取到相应的消息，保证了消息消费的可靠性。- 消息回溯 - Consumer已经消费成功的消息，或者之前消费业务逻辑有问题，现在需要重新消费。要支持此功能，则Broker存储端在向Consumer消费端投递成功消息后，消息仍然需要保留。重新消费一般是按照时间维度，例如由于Consumer系统故障，恢复后需要重新消费1小时前的数据。RocketMQ Broker提供了一种机制，可以按照时间维度来回退消费进度，这样就可以保证只要发送成功的消息，只要消息没有过期，消息始终是可以消费到的。

重复消费

消费者端做幂等

如何解决消息堆积

产生原因：生产者生产消息过快or消费者消费消息太慢

生产者限流降级，增加多个消费者实例。