Kafka篇

Why kafka

消息队列的作用：异步、削峰填谷、解耦

中小型公司，技术实力较为一般，技术挑战不是特别高，用 RabbitMQ （开源、社区活跃）是不错的选择；大型公司，基础架构研发实力较强，用 RocketMQ（Java二次开发） 是很好的选择。

如果是大数据领域的实时计算、日志采集等场景，用 Kafka 是业内标准的，绝对没问题，社区活跃度很高，绝对不会黄，何况几乎是全世界这个领域的事实性规范。

img

RabbitMQ

RabbitMQ开始是用在电信业务的可靠通信的，也是少有的几款支持AMQP协议的产品之一。

优点：

  • 轻量级，快速，部署使用方便
  • 支持灵活的路由配置。RabbitMQ中，在生产者和队列之间有一个交换器模块。根据配置的路由规则，生产者发送的消息可以发送到不同的队列中。路由规则很灵活，还可以自己实现。
  • RabbitMQ的客户端支持大多数的编程语言，支持AMQP协议。

缺点：

  • 如果有大量消息堆积在队列中，性能会急剧下降
  • 每秒处理几万到几十万的消息。如果应用要求高的性能，不要选择RabbitMQ。
  • RabbitMQ是Erlang开发的，功能扩展和二次开发代价很高。

RocketMQ

借鉴了Kafka的设计并做了很多改进，几乎具备了消息队列应该具备的所有特性和功能。

  • RocketMQ主要用于有序，事务，流计算，消息推送，日志流处理，binlog分发等场景。
  • 经过了历次的双11考验，性能，稳定性可靠性没的说。
  • java开发，阅读源代码、扩展、二次开发很方便。
  • 对电商领域的响应延迟做了很多优化。
  • 每秒处理几十万的消息，同时响应在毫秒级。如果应用很关注响应时间，可以使用RocketMQ。
  • 性能比RabbitMQ高一个数量级，。
  • 支持死信队列，DLX 是一个非常有用的特性。它可以处理异常情况下，消息不能够被消费者正确消费而被置入死信队列中的情况，后续分析程序可以通过消费这个死信队列中的内容来分析当时所遇到的异常情况，进而可以改善和优化系统。

缺点：

跟周边系统的整合和兼容不是很好。

Kafka

高可用，几乎所有相关的开源软件都支持，满足大多数的应用场景，尤其是大数据和流计算领域，

  • Kafka高效，可伸缩，消息持久化。支持分区、副本和容错。
  • 对批处理和异步处理做了大量的设计，因此Kafka可以得到非常高的性能。
  • 每秒处理几十万异步消息消息，如果开启了压缩，最终可以达到每秒处理2000w消息的级别。
  • 但是由于是异步的和批处理的，延迟也会高，不适合电商场景。

What Kafka

  • Producer API：允许应用程序将记录流发布到一个或多个Kafka主题。
  • Consumer API：允许应用程序订阅一个或多个主题并处理为其生成的记录流。
  • Streams API：允许应用程序充当流处理器，将输入流转换为输出流。

img

消息Message

Kafka的数据单元称为消息。可以把消息看成是数据库里的一个“数据行”或一条“记录”。

批次

为了提高效率，消息被分批写入Kafka。提高吞吐量却加大了响应时间

主题Topic

通过主题进行分类，类似数据库中的表，

分区Partition

Topic可以被分成若干分区分布于kafka集群中，方便扩容

单个分区内是有序的，partition设置为一才能保证全局有序

副本Replicas

每个主题被分为若干个分区，每个分区有多个副本。

生产者Producer

生产者在默认情况下把消息均衡地分布到主题的所有分区上：

  • 直接指定消息的分区
  • 根据消息的key散列取模得出分区
  • 轮询指定分区。

消费者Comsumer

消费者通过偏移量来区分已经读过的消息，从而消费消息。把每个分区最后读取的消息偏移量保存在Zookeeper 或Kafka上，如果消费者关闭或重启，它的读取状态不会丢失。

消费组ComsumerGroup

消费组保证每个分区只能被一个消费者使用，避免重复消费。如果群组内一个消费者失效，消费组里的其他消费者可以接管失效消费者的工作再平衡，重新分区

节点Broker

连接生产者和消费者，单个broker可以轻松处理数千个分区以及每秒百万级的消息量。

  • broker接收来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保存。
  • broker为消费者提供服务，响应读取分区的请求，返回已经提交到磁盘上的消息。

集群

每隔分区都有一个首领，当分区被分配给多个broker时，会通过首领进行分区复制。

生产者Offset

消息写入的时候，每一个分区都有一个offset，即每个分区的最新最大的offset。

消费者Offset

不同消费组中的消费者可以针对一个分区存储不同的Offset，互不影响

LogSegment

  • 一个分区由多个LogSegment组成，
  • 一个LogSegment由.log .index .timeindex组成
  • .log追加是顺序写入的，文件名是以文件中第一条message的offset来命名的
  • .Index进行日志删除的时候和数据查找的时候可以快速定位。
  • .timeStamp则根据时间戳查找对应的偏移量。

How Kafka

优点

  • 高吞吐量：单机每秒处理几十上百万的消息量。即使存储了TB及消息，也保持稳定的性能。

    • 零拷贝 减少内核态到用户态的拷贝，磁盘通过sendfile实现DMA 拷贝Socket buffer
    • 顺序读写 充分利用磁盘顺序读写的超高性能
    • 页缓存mmap，将磁盘文件映射到内存, 用户通过修改内存就能修改磁盘文件。

  • 高性能：单节点支持上千个客户端，并保证零停机和零数据丢失。

  • 持久化：将消息持久化到磁盘。通过将数据持久化到硬盘以及replication防止数据丢失。

  • 分布式系统，易扩展。所有的组件均为分布式的，无需停机即可扩展机器。

  • 可靠性 - Kafka是分布式，分区，复制和容错的。

  • 客户端状态维护：消息被处理的状态是在Consumer端维护，当失败时能自动平衡。

应用场景

  • 日志收集：用Kafka可以收集各种服务的Log，通过大数据平台进行处理；
  • 消息系统：解耦生产者和消费者、缓存消息等；
  • 用户活动跟踪：Kafka经常被用来记录Web用户或者App用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到Kafka的Topic中，然后消费者通过订阅这些Topic来做运营数据的实时的监控分析，也可保存到数据库；

生产消费基本流程

  1. Producer创建时，会创建一个Sender线程并设置为守护线程。
  2. 生产的消息先经过拦截器->序列化器->分区器，然后将消息缓存在缓冲区。
  3. 批次发送的条件为：缓冲区数据大小达到batch.size或者linger.ms达到上限。
  4. 批次发送后，发往指定分区，然后落盘到broker；
    • acks=0只要将消息放到缓冲区，就认为消息已经发送完成。
    • acks=1表示消息只需要写到主分区即可。在该情形下，如果主分区收到消息确认之后就宕机了，而副本分区还没来得及同步该消息，则该消息丢失。
    • acks=all （默认）首领分区会等待所有的ISR副本分区确认记录。该处理保证了只要有一个ISR副本分区存活，消息就不会丢失。
  1. 如果生产者配置了retrires参数大于0并且未收到确认，那么客户端会对该消息进行重试。
  2. 落盘到broker成功，返回生产元数据给生产者。

Leader选举

  • Kafka会在Zookeeper上针对每个Topic维护一个称为ISR（in-sync replica）的集合

  • 当集合中副本都跟Leader中的副本同步了之后，kafka才会认为消息已提交

  • 只有这些跟Leader保持同步的Follower才应该被选作新的Leader

  • 假设某个topic有N+1个副本，kafka可以容忍N个服务器不可用，冗余度较低如果ISR中的副本都丢失了，则：

    • 可以等待ISR中的副本任何一个恢复，接着对外提供服务，需要时间等待
    • 从OSR中选出一个副本做Leader副本，此时会造成数据丢失

副本消息同步

首先，Follower 发送 FETCH 请求给 Leader。接着，Leader 会读取底层日志文件中的消 息数据，再更新它内存中的 Follower 副本的 LEO 值，更新为 FETCH 请求中的 fetchOffset 值。最后，尝试更新分区高水位值。Follower 接收到 FETCH 响应之后，会把消息写入到底层日志，接着更新 LEO 和 HW 值。

相关概念：LEO和HW。

  • LEO：即日志末端位移(log end offset)，记录了该副本日志中下一条消息的位移值。如果LEO=10，那么表示该副本保存了10条消息，位移值范围是[0, 9]
  • HW：水位值HW（high watermark）即已备份位移。对于同一个副本对象而言，其HW值不会大于LEO值。小于等于HW值的所有消息都被认为是“已备份”的（replicated）

Rebalance

  • 组成员数量发生变化
  • 订阅主题数量发生变化
  • 订阅主题的分区数发生变化

leader选举完成后，当以上三种情况发生时，Leader根据配置的RangeAssignor开始分配消费方案，即哪个consumer负责消费哪些topic的哪些partition。一旦完成分配，leader会将这个方案封装进SyncGroup请求中发给coordinator，非leader也会发SyncGroup请求，只是内容为空。coordinator接收到分配方案之后会把方案塞进SyncGroup的response中发给各个consumer。这样组内的所有成员就都知道自己应该消费哪些分区了。

分区分配算法RangeAssignor

  • 原理是按照消费者总数和分区总数进行整除运算平均分配给所有的消费者。
  • 订阅Topic的消费者按照名称的字典序排序，分均分配，剩下的字典序从前往后分配

增删改查

如何查看偏移量为23的消息？

通过查询跳跃表ConcurrentSkipListMap，定位到在00000000000000000000.index ，通过二分法在偏移量索引文件中找到不大于 23 的最大索引项，即offset 20 那栏，然后从日志分段文件中的物理位置为320 开始顺序查找偏移量为 23 的消息。

切分文件

  • 大小分片 当前日志分段文件的大小超过了 broker 端参数 log.segment.bytes 配置的值
  • 时间分片 当前日志分段中消息的最大时间戳与系统的时间戳的差值大于log.roll.ms配置的值
  • 索引分片 偏移量或时间戳索引文件大小达到broker端 log.index.size.max.bytes配置的值
  • 偏移分片 追加的消息的偏移量与当前日志分段的偏移量之间的差值大于 Integer.MAX_VALUE

一致性

幂等性

保证在消息重发的时候，消费者不会重复处理。即使在消费者收到重复消息的时候，重复处理，也

要保证最终结果的一致性。所谓幂等性，数学概念就是： f(f(x)) = f(x)

如何实现？

添加唯一ID，类似于数据库的主键，用于唯一标记一个消息。

如何选举

  1. 使用 Zookeeper 的分布式锁选举控制器，并在节点加入集群或退出集群时通知控制器。
  2. 控制器负责在节点加入或离开集群时进行分区Leader选举。
  3. 控制器使用epoch忽略小的纪元来避免脑裂：两个节点同时认为自己是当前的控制器。

可用性

  • 创建Topic的时候可以指定 –replication-factor 3 ，表示不超过broker的副本数
  • 只有Leader是负责读写的节点，Follower定期地到Leader上Pull数据。
  • ISR是Leader负责维护的与其保持同步的Replica列表，即当前活跃的副本列表。如果一个Follow落后太多，Leader会将它从ISR中移除。选举时优先从ISR中挑选Follower。
  • 设置 acks=all 。Leader收到了ISR中所有Replica的ACK，才向Producer发送ACK。

面试题

1、线上问题rebalance

因集群架构变动导致的消费组内重平衡，如果kafka集内节点较多，比如数百个，那重平衡可能会耗时导致数分钟到数小时，此时kafka基本处于不可用状态，对kafka的TPS影响极大

产生的原因：

  • 组成员数量发生变化
  • 订阅主题数量发生变化
  • 订阅主题的分区数发生变化组成员崩溃和组成员主动离开是两个不同的场景。因为在崩溃时成员并不会主动地告知coordinator此事，coordinator有可能需要一个完整的session.timeout周期(心跳周期)才能检测到这种崩溃，这必然会造成consumer的滞后。可以说离开组是主动地发起rebalance；而崩溃则是被动地发起rebalance。

解决方案：

  1. 加大超时时间 session.timout.ms=6s
  2. 加大心跳频率 heartbeat.interval.ms=2s
  3. 增长推送间隔 max.poll.interval.ms=t+1 minutes

第一类非必要 Rebalance 是因为未能及时发送心跳，导致 Consumer 被“踢出”Group 而引发的。
因此，你需要仔细地设置 session.timeout.ms 和 heartbeat.interval.ms 的值。我在这里给出一些推荐数值，你可以“无脑”地应用在你的生产环境中。

  • 设置 session.timeout.ms = 6s。
  • 设置 heartbeat.interval.ms = 2s。
  • 要保证 Consumer 实例在被判定为“dead”之前，能够发送至少 3 轮的心跳请求，即 session.timeout.ms >= 3 * heartbeat.interval.ms

第二类非必要 Rebalance 是 Consumer 消费时间过长导致的。增大max.poll.interval.ms 参数值。

2、ZooKeeper 的作用

目前，Kafka 使用 ZooKeeper 存放集群元数据、成员管理、Controller 选举，以及其他一些管理类任务。之后，等 KIP-500 提案完成后，Kafka 将完全不再依赖于 ZooKeeper。

  • 存放元数据是指主题分区的所有数据都保存在 ZooKeeper 中，其他“人”都要与它保持对齐。
  • 成员管理是指 Broker 节点的注册、注销以及属性变更等 。
  • Controller 选举是指选举集群 Controller，包括但不限于主题删除、参数配置等。

一言以蔽之:KIP-500 ，是使用社区自研的基于 Raft 的共识算法，实现 Controller 自选举。

同样是存储元数据，这几年基于Raft算法的etcd认可度越来越高

越来越多的系统开始用它保存关键数据。比如，秒杀系统经常用它保存各节点信息，以便控制消费 MQ 的服务数量。还有些业务系统的配置数据，也会通过 etcd 实时同步给业务系统的各节点，比如，秒杀管理后台会使用 etcd 将秒杀活动的配置数据实时同步给秒杀 API 服务各节点。

3、Replica副本的作用

Kafka 只有 Leader 副本才能 对外提供读写服务，响应 Clients 端的请求。Follower 副本只是采用拉(PULL)的方 式，被动地同步 Leader 副本中的数据，并且在 Leader 副本所在的 Broker 宕机后，随时准备应聘 Leader 副本。

  • 自 Kafka 2.4 版本开始，社区可以通过配置参数，允许 Follower 副本有限度地提供读服务。
  • 之前确保一致性的主要手段是高水位机制， 但高水位值无法保证 Leader 连续变更场景下的数据一致性，因此，社区引入了 Leader Epoch 机制，来修复高水位值的弊端。

4、为什么不支持读写分离?

  • 自 Kafka 2.4 之后，Kafka 提供了有限度的读写分离。
  • 场景不适用。读写分离适用于那种读负载很大，而写操作相对不频繁的场景。
  • 同步机制。Kafka 采用 PULL 方式实现 Follower 的同步，同时复制延迟较大。

5、如何防止重复消费

  • 代码层面每次消费需提交offset
  • 通过Mysql的唯一键约束，结合Redis查看id是否被消费，存Redis可以直接使用set方法
  • 量大且允许误判的情况下，使用布隆过滤器也可以

6、如何保证数据不会丢失

  • 生产者生产消息可以通过comfirm配置ack=all解决
  • Broker同步过程中leader宕机可以通过配置ISR副本+重试解决
  • 消费者丢失可以关闭自动提交offset功能，系统处理完成时提交offset

7、如何保证顺序消费

  • 单 topic，单partition，单 consumer，单线程消费，吞吐量低，不推荐
  • 如只需保证单key有序，为每个key申请单独内存 queue，每个线程分别消费一个内存 queue 即可，这样就能保证单key（例如用户id、活动id）顺序性。

8、线上如何解决积压消费

  • 修复consumer，使其具备消费能力，并且扩容N台
  • 写一个分发的程序，将Topic均匀分发到临时Topic中
  • 同时起N台consumer，消费不同的临时Topic

9、如何避免消息积压

  • 提高消费并行度
  • 批量消费
  • 减少组件IO的交互次数
  • 优先级消费

10、如何设计消息队列

需要支持快速水平扩容，broker+partition，partition放不同的机器上，增加机器时将数据根据topic做迁移，分布式需要考虑一致性、可用性、分区容错性

  • 一致性：生产者的消息确认、消费者的幂等性、Broker的数据同步
  • 可用性：数据如何保证不丢不重、数据如何持久化、持久化时如何读写
  • 分区容错：采用何种选举机制、如何进行多副本同步
  • 海量数据：如何解决消息积压、海量Topic性能下降

性能上，可以借鉴时间轮、零拷贝、IO多路复用、顺序读写、压缩批处理

MQ常见面试题

1、你们为什么使用mq？具体的使用场景是什么？

mq的作用很简单，削峰填谷。以电商交易下单的场景来说，正向交易的过程可能涉及到创建订单、扣减库存、扣减活动预算、扣减积分等等。每个接口的耗时如果是100ms，那么理论上整个下单的链路就需要耗费400ms，这个时间显然是太长了。

img

如果这些操作全部同步处理的话，首先调用链路太长影响接口性能，其次分布式事务的问题很难处理，这时候像扣减预算和积分这种对实时一致性要求没有那么高的请求，完全就可以通过mq异步的方式去处理了。同时，考虑到异步带来的不一致的问题，我们可以通过job去重试保证接口调用成功，而且一般公司都会有核对的平台，比如下单成功但是未扣减积分的这种问题可以通过核对作为兜底的处理方案。

img

使用mq之后我们的链路变简单了，同时异步发送消息我们的整个系统的抗压能力也上升了。

2、那你们使用什么mq？基于什么做的选型？

我们主要调研了几个主流的mq，kafka、rabbitmq、rocketmq、activemq，选型我们主要基于以下几个点去考虑：

  1. 由于我们系统的qps压力比较大，所以性能是首要考虑的要素。
  2. 开发语言，由于我们的开发语言是java，主要是为了方便二次开发。
  3. 对于高并发的业务场景是必须的，所以需要支持分布式架构的设计。
  4. 功能全面，由于不同的业务场景，可能会用到顺序消息、事务消息等。

基于以上几个考虑，我们最终选择了RocketMQ。

img

3、你上面提到异步发送，那消息可靠性怎么保证？

消息丢失可能发生在生产者发送消息、MQ本身丢失消息、消费者丢失消息3个方面。

生产者丢失

生产者丢失消息的可能点在于程序发送失败抛异常了没有重试处理，或者发送的过程成功但是过程中网络闪断MQ没收到，消息就丢失了。

由于同步发送的一般不会出现这样使用方式，所以我们就不考虑同步发送的问题，我们基于异步发送的场景来说。

异步发送分为两个方式：异步有回调和异步无回调，无回调的方式，生产者发送完后不管结果可能就会造成消息丢失，而通过异步发送+回调通知+本地消息表的形式我们就可以做出一个解决方案。以下单的场景举例。

  1. 下单后先保存本地数据和MQ消息表，这时候消息的状态是发送中，如果本地事务失败，那么下单失败，事务回滚。
  2. 下单成功，直接返回客户端成功，异步发送MQ消息
  3. MQ回调通知消息发送结果，对应更新数据库MQ发送状态
  4. JOB轮询超过一定时间（时间根据业务配置）还未发送成功的消息去重试
  5. 在监控平台配置或者JOB程序处理超过一定次数一直发送不成功的消息，告警，人工介入。

img
一般而言，对于大部分场景来说异步回调的形式就可以了，只有那种需要完全保证不能丢失消息的场景我们做一套完整的解决方案。

MQ丢失

如果生产者保证消息发送到MQ，而MQ收到消息后还在内存中，这时候宕机了又没来得及同步给从节点，就有可能导致消息丢失。

比如RocketMQ：

RocketMQ分为同步刷盘和异步刷盘两种方式，默认的是异步刷盘，就有可能导致消息还未刷到硬盘上就丢失了，可以通过设置为同步刷盘的方式来保证消息可靠性，这样即使MQ挂了，恢复的时候也可以从磁盘中去恢复消息。

比如Kafka也可以通过配置做到：

1
2
3
4
acks=all 只有参与复制的所有节点全部收到消息，才返回生产者成功。这样的话除非所有的节点都挂了，消息才会丢失。
replication.factor=N,设置大于1的数，这会要求每个partion至少有2个副本
min.insync.replicas=N，设置大于1的数，这会要求leader至少感知到一个follower还保持着连接
retries=N，设置一个非常大的值，让生产者发送失败一直重试

虽然我们可以通过配置的方式来达到MQ本身高可用的目的，但是都对性能有损耗，怎样配置需要根据业务做出权衡。

消费者丢失

消费者丢失消息的场景：消费者刚收到消息，此时服务器宕机，MQ认为消费者已经消费，不会重复发送消息，消息丢失。

RocketMQ默认是需要消费者回复ack确认，而kafka需要手动开启配置关闭自动offset。

消费方不返回ack确认，重发的机制根据MQ类型的不同发送时间间隔、次数都不尽相同，如果重试超过次数之后会进入死信队列，需要手工来处理了。（Kafka没有这些）

img

4、你说到消费者消费失败的问题，那么如果一直消费失败导致消息积压怎么处理？

因为考虑到时消费者消费一直出错的问题，那么我们可以从以下几个角度来考虑：

  1. 消费者出错，肯定是程序或者其他问题导致的，如果容易修复，先把问题修复，让consumer恢复正常消费
  2. 如果时间来不及处理很麻烦，做转发处理，写一个临时的consumer消费方案，先把消息消费，然后再转发到一个新的topic和MQ资源，这个新的topic的机器资源单独申请，要能承载住当前积压的消息
  3. 处理完积压数据后，修复consumer，去消费新的MQ和现有的MQ数据，新MQ消费完成后恢复原状

img

5、那如果消息积压达到磁盘上限，消息被删除了怎么办？

最初，我们发送的消息记录是落库保存了的，而转发发送的数据也保存了，那么我们就可以通过这部分数据来找到丢失的那部分数据，再单独跑个脚本重发就可以了。如果转发的程序没有落库，那就和消费方的记录去做对比，只是过程会更艰难一点。

6、说了这么多，那你说说RocketMQ实现原理吧？

RocketMQ由NameServer注册中心集群、Producer生产者集群、Consumer消费者集群和若干Broker（RocketMQ进程）组成，它的架构原理是这样的：

  1. Broker在启动的时候去向所有的NameServer注册，并保持长连接，每30s发送一次心跳
  2. Producer在发送消息的时候从NameServer获取Broker服务器地址，根据负载均衡算法选择一台服务器来发送消息
  3. Conusmer消费消息的时候同样从NameServer获取Broker地址，然后主动拉取消息来消费

img

7、为什么RocketMQ不使用Zookeeper作为注册中心呢？

我认为有以下几个点是不使用zookeeper的原因：

  1. 根据CAP理论，同时最多只能满足两个点，而zookeeper满足的是CP，也就是说zookeeper并不能保证服务的可用性，zookeeper在进行选举的时候，整个选举的时间太长，期间整个集群都处于不可用的状态，而这对于一个注册中心来说肯定是不能接受的，作为服务发现来说就应该是为可用性而设计。
  2. 基于性能的考虑，NameServer本身的实现非常轻量，而且可以通过增加机器的方式水平扩展，增加集群的抗压能力，而zookeeper的写是不可扩展的，而zookeeper要解决这个问题只能通过划分领域，划分多个zookeeper集群来解决，首先操作起来太复杂，其次这样还是又违反了CAP中的A的设计，导致服务之间是不连通的。
  3. 持久化的机制来带的问题，ZooKeeper 的 ZAB 协议对每一个写请求，会在每个 ZooKeeper 节点上保持写一个事务日志，同时再加上定期的将内存数据镜像（Snapshot）到磁盘来保证数据的一致性和持久性，而对于一个简单的服务发现的场景来说，这其实没有太大的必要，这个实现方案太重了。而且本身存储的数据应该是高度定制化的。
  4. 消息发送应该弱依赖注册中心，而RocketMQ的设计理念也正是基于此，生产者在第一次发送消息的时候从NameServer获取到Broker地址后缓存到本地，如果NameServer整个集群不可用，短时间内对于生产者和消费者并不会产生太大影响。

8、那Broker是怎么保存数据的呢？

RocketMQ主要的存储文件包括commitlog文件、consumequeue文件、indexfile文件。

Broker在收到消息之后，会把消息保存到commitlog的文件当中，而同时在分布式的存储当中，每个broker都会保存一部分topic的数据，同时，每个topic对应的messagequeue下都会生成consumequeue文件用于保存commitlog的物理位置偏移量offset，indexfile中会保存key和offset的对应关系。

img
CommitLog文件保存于${Rocket_Home}/store/commitlog目录中，从图中我们可以明显看出来文件名的偏移量，每个文件默认1G，写满后自动生成一个新的文件。

img
由于同一个topic的消息并不是连续的存储在commitlog中，消费者如果直接从commitlog获取消息效率非常低，所以通过consumequeue保存commitlog中消息的偏移量的物理地址，这样消费者在消费的时候先从consumequeue中根据偏移量定位到具体的commitlog物理文件，然后根据一定的规则（offset和文件大小取模）在commitlog中快速定位。

img

9、Master和Slave之间是怎么同步数据的呢？

而消息在master和slave之间的同步是根据raft协议来进行的：

  1. 在broker收到消息后，会被标记为uncommitted状态
  2. 然后会把消息发送给所有的slave
  3. slave在收到消息之后返回ack响应给master
  4. master在收到超过半数的ack之后，把消息标记为committed
  5. 发送committed消息给所有slave，slave也修改状态为committed

10、你知道RocketMQ为什么速度快吗？

是因为使用了顺序存储、Page Cache和异步刷盘。

  1. 我们在写入commitlog的时候是顺序写入的，这样比随机写入的性能就会提高很多
  2. 写入commitlog的时候并不是直接写入磁盘，而是先写入操作系统的PageCache
  3. 最后由操作系统异步将缓存中的数据刷到磁盘

11、什么是事务、半事务消息？怎么实现的？

事务消息就是MQ提供的类似XA的分布式事务能力，通过事务消息可以达到分布式事务的最终一致性。

半事务消息就是MQ收到了生产者的消息，但是没有收到二次确认，不能投递的消息。

实现原理如下：

  1. 生产者先发送一条半事务消息到MQ
  2. MQ收到消息后返回ack确认
  3. 生产者开始执行本地事务
  4. 如果事务执行成功发送commit到MQ，失败发送rollback
  5. 如果MQ长时间未收到生产者的二次确认commit或者rollback，MQ对生产者发起消息回查
  6. 生产者查询事务执行最终状态
  7. 根据查询事务状态再次提交二次确认

最终，如果MQ收到二次确认commit，就可以把消息投递给消费者，反之如果是rollback，消息会保存下来并且在3天后被删除。

img

快来参与讨论吧~


本站总访问量次