消息中间件架构体系 - Kafka：从“入门”到“进阶”，也不是

1.0 前言

Kafka 是一款分布式消息发布和订阅系统，具有高性能、高吞吐量的特点而被广泛应用与大数据传输场景。它是由 LinkedIn 公司开发，使用 Scala 语言编写，之后成为 Apache 基金会的一个顶级项目。kafka 提供了类似 JMS 的特性，但是在设计和实现上是完全不同的，而且他也不是 JMS 规范的实现。

2.0 Kafka简介

2.1 kafka产生背景

kafka 作为一个消息系统，早起设计的目的是用作 LinkedIn 的活动流（Activity Stream）和运营数据处理管道（Pipeline）。活动流数据是所有的网站对用户的使用情况做分析的时候要用到的最常规的部分,活动数据包括页面的访问量（PV）、被查看内容方面的信息以及搜索内容。这种数据通常的处理方式是先把各种活动以日志的形式写入某种文件，然后周期性的对这些文件进行统计分析。运营数据指的是服务器的性能数据（CPU、IO 使用率、请求时间、服务日志等）。

2.2 Kafka应用场景

由于 kafka 具有更好的吞吐量、内置分区、冗余及容错性的优点(kafka 每秒可以处理几十万消息)，让 kafka 成为了一个很好的大规模消息处理应用的解决方案。

所以在企业级应用长，主要会应用于如下几个方面

行为跟踪：kafka 可以用于跟踪用户浏览页面、搜索及其他行为。通过发布-订阅模式实时记录到对应的 topic中，通过后端大数据平台接入处理分析，并做更进一步的实时处理和监控

日志收集：日志收集方面，有很多比较优秀的产品，比如 Apache Flume，很多公司使用kafka 代理日志聚合。日志聚合表示从服务器上收集日志文件，然后放到一个集中的平台（文件服务器）进行处理。在实际应用开发中，我们应用程序的 log 都会输出到本地的磁盘上，排查问题的话通过 linux 命令来搞定，如果应用程序组成了负载均衡集群，并且集群的机器有几十台以上，那么想通过日志快速定位到问题，就是很麻烦的事情了。所以一般都会做一个日志统一收集平台管理 log 日志用来快速查询重要应用的问题。所以很多公司的套路都是把应用日志几种到 kafka 上，然后分别导入到 es 和 hdfs 上，用来做实时检索分析和离线统计数据备份等。而另一方面，kafka 本身又提供了很好的 api 来集成日志并且做日志收集

2.3 kafka架构

一个典型的 kafka 集群包含若干 Producer（可以是应用节点产生的消息，也可以是通过Flume 收集日志产生的事件），若干个 Broker（kafka 支持水平扩展）、若干个 Consumer Group，以及一个 zookeeper 集群。kafka 通过 zookeeper 管理集群配置及服务协同。

Producer 使用 push 模式将消息发布到 broker，consumer 通过监听使用 pull 模式从broker 订阅并消费消息。多个 broker 协同工作，producer 和 consumer 部署在各个业务逻辑中。三者通过zookeeper 管理协调请求和转发。这样就组成了一个高性能的分布式消息发布和订阅系统。图上有一个细节是和其他 mq 中间件不同的点，producer 发送消息到 broker的过程是 push，而 consumer 从 broker 消费消息的过程是 pull，主动去拉数据。而不是 broker 把数据主动发送给 consumer

名词解释:

Topic

Kafka将消息分门别类，每一类的消息称之为一个主题（Topic）。

Producer

发布消息的对象称之为主题生产者（Kafka topic producer）

Consumer

订阅消息并处理发布的消息的对象称之为主题消费者（consumers）

Broker

已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理（Broker）。 消费者可以订阅一个或多个主题（topic），并从Broker拉数据，从而消费这些已发布的消息。

2.4 Topic和Log

Topic是发布的消息的类别名，一个topic可以有零个，一个或多个消费者订阅该主题的消息。

对于每个topic，Kafka集群都会维护一个分区log，就像下图中所示：

每一个分区都是一个顺序的、不可变的消息队列， 并且可以持续的添加。分区中的消息都被分了一个序列号，称之为偏移量(offset)，在每个分区中此偏移量都是唯一的。

Kafka集群保持所有的消息，直到它们过期（无论消息是否被消费）。实际上消费者所持有的仅有的元数据就是这个offset（偏移量），也就是说offset由消费者来控制：正常情况当消费者消费消息的时候，偏移量也线性的的增加。但是实际偏移量由消费者控制，消费者可以将偏移量重置为更早的位置，重新读取消息。可以看到这种设计对消费者来说操作自如，一个消费者的操作不会影响其它消费者对此log的处理。

再说说分区。Kafka中采用分区的设计有几个目的。一是可以处理更多的消息，不受单台服务器的限制。Topic拥有多个分区意味着它可以不受限的处理更多的数据。第二，分区可以作为并行处理的单元，稍后会谈到这一点。

2.5 分布式

Log的分区被分布到集群中的多个服务器上。每个服务器处理它分到的分区。 根据配置每个分区还可以复制到其它服务器作为备份容错。 每个分区有一个leader，零或多个follower。Leader处理此分区的所有的读写请求，而follower被动的复制数据。如果leader宕机，其它的一个follower会被推举为新的leader。 一台服务器可能同时是一个分区的leader，另一个分区的follower。 这样可以平衡负载，避免所有的请求都只让一台或者某几台服务器处理。

2.6 生产者

生产者往某个Topic上发布消息。生产者也负责选择发布到Topic上的哪一个分区。最简单的方式从分区列表中轮流选择。也可以根据某种算法依照权重选择分区。开发者负责如何选择分区的算法。

2.7 消费者

通常来讲，消息模型可以分为两种， 队列和发布-订阅式。 队列的处理方式是 一组消费者从服务器读取消息，一条消息只有其中的一个消费者来处理。在发布-订阅模型中，消息被广播给所有的消费者，接收到消息的消费者都可以处理此消息。Kafka为这两种模型提供了单一的消费者抽象模型： 消费者组 （consumer group）。 消费者用一个消费者组名标记自己。 一个发布在Topic上消息被分发给此消费者组中的一个消费者。 假如所有的消费者都在一个组中，那么这就变成了queue模型。 假如所有的消费者都在不同的组中，那么就完全变成了发布-订阅模型。 更通用的， 我们可以创建一些消费者组作为逻辑上的订阅者。每个组包含数目不等的消费者， 一个组内多个消费者可以用来扩展性能和容错。正如下图所示：

2个kafka集群托管4个分区（P0-P3），2个消费者组，消费组A有2个消费者实例，消费组B有4个。

正像传统的消息系统一样，Kafka保证消息的顺序不变。 再详细扯几句。传统的队列模型保持消息，并且保证它们的先后顺序不变。但是， 尽管服务器保证了消息的顺序，消息还是异步的发送给各个消费者，消费者收到消息的先后顺序不能保证了。这也意味着并行消费将不能保证消息的先后顺序。用过传统的消息系统的同学肯定清楚，消息的顺序处理很让人头痛。如果只让一个消费者处理消息，又违背了并行处理的初衷。 在这一点上Kafka做的更好，尽管并没有完全解决上述问题。 Kafka采用了一种分而治之的策略：分区。 因为Topic分区中消息只能由消费者组中的唯一一个消费者处理，所以消息肯定是按照先后顺序进行处理的。但是它也仅仅是保证Topic的一个分区顺序处理，不能保证跨分区的消息先后处理顺序。 所以，如果你想要顺序的处理Topic的所有消息，那就只提供一个分区。

3.0 Docker搭建kafka

3.1 下载以下三个镜像

dockerpull wurstmeister/zookeeperdockerpull wurstmeister/kafkadockerpull sheepkiller/kafka-manager

kafka-manager是kafka的可视化管理工具

3.2 后动容器

dockerrun -d --name zookeeper --publish2181:2181\--volume /etc/localtime:/etc/localtime \--restart=always \wurstmeister/zookeeperdocker run -d --name kafka --publish9082:9092\--link zookeeper:zookeeper \--env KAFKA_BROKER_ID=100\--env HOST_IP=127.0.0.1\--env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181\--env KAFKA_ADVERTISED_HOST_NAME=192.168.1.108\--env KAFKA_ADVERTISED_PORT=9082\--restart=always \--volume /etc/localtime:/etc/localtime \wurstmeister/kafkadocker run -d --name kafka-manager \--link zookeeper:zookeeper \--link kafka:kafka -p9001:9000\--restart=always \--env ZK_HOSTS=zookeeper:2181\sheepkiller/kafka-manager

3.3 访问

http://127.0.0.1:9001

3.4 添加Cluster

3.5 查看界面

搭建完毕,页面其他功能自己摸索下

4.0 Kafka快速入门

//以下Spring Boot应用程序将三个消息发送到一个主题，接收它们，然后停止：

@SpringBootApplicationpublicclassApplicationimplementsCommandLineRunner{publicstaticLogger logger = LoggerFactory.getLogger(Application.class);publicstaticvoidmain(String[] args){        SpringApplication.run(Application.class,args).close();    }@AutowiredprivateKafkaTemplate template;privatefinalCountDownLatch latch =newCountDownLatch(3);@Overridepublicvoidrun(String... args)throwsException{this.template.send("myTopic","foo1");this.template.send("myTopic","foo2");this.template.send("myTopic","foo3");        latch.await(60, TimeUnit.SECONDS);        logger.info("All received");    }@KafkaListener(topics ="myTopic")publicvoidlisten(ConsumerRecord<?, ?> cr)throwsException{        logger.info(cr.toString());        latch.countDown();    }}

5.0 Kafka进阶

5.1 通信原理

消息是 kafka 中最基本的数据单元，在 kafka 中，一条消息由 key、 value 两部分构成，在发送一条消息时，我们可以指定这个 key，那么 producer 会根据 key 和 partition 机制来判断当前这条消息应该发送并存储到哪个 partition 中。我们可以根据需要进行扩展 producer 的 partition 机制。

默认情况下，kafka 采用的是 hash 取模的分区算法。如果Key 为 null，则会随机分配一个分区。这个随机是在这个参数”metadata.max.age.ms”的时间范围内随机选择一个。对于这个时间段内，如果 key 为 null，则只会发送到唯一的分区。这个值值哦默认情况下是 10 分钟更新一次。

关于 Metadata ，这个之前没讲过，简单理解就是T opic/Partition 和 broker 的映射关系，每一个 topic 的每一个 partition，需要知道对应的 broker 列表是什么， leader是谁、 follower 是谁。这些信息都是存储在 Metadata 这个类里面。

5.1.1 消费端如何消费指定的分区

//通过下面的代码，就可以消费指定该 topic 下的 0 号分区。其他分区的数据就无法接收//消费指定分区的时候，不需要再订阅//kafkaConsumer.subscribe(Collections.singletonList(topic));//消费指定的分区TopicPartition topicPartition=newTopicPartition(topic,0);kafkaConsumer.assign(Arrays.asList(topicPartition));

5.1.2 消费原理

在实际生产过程中，每个 topic 都会有多个 partitions，多个 partitions 的好处在于，一方面能够对 broker 上的数据进行分片有效减少了消息的容量从而提升 io 性能。另外一方面，为了提高消费端的消费能力，一般会通过多个consumer 去消费同一个 topic ，也就是消费端的负载均衡机制，也就是我们接下来要了解的，在多个partition 以及多个 consumer 的情况下，消费者是如何消费消息的同时，在上一节课，我们讲了， kafka 存在 consumer group的 概 念 ， 也 就是 group.id 一样 的 consumer ，这些consumer 属于一个 consumer group，组内的所有消费者协调在一起来消费订阅主题的所有分区。当然每一个分区只能由同一个消费组内的 consumer 来消费，那么同一个consumer group 里面的 consumer 是怎么去分配该消费哪个分区里的数据的呢？如下图所示， 3 个分区， 3 个消费者，那么哪个消费者消分哪个分区？

分区分配策略

在 kafka 中，存在两种分区分配策略，一种是 Range(默认)、另一种还 是 RoundRobin （ 轮 询 ）。 通过partition.assignment.strategy 这个参数来设置。

Range strategy（范围分区）

Range 策略是对每个主题而言的，首先对同一个主题里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。假设我们有 10 个分区，3 个消费者，排完序的分区将会是 0, 1, 2, 3, 4, 5, 6, 7, 8, 9；消费者线程排完序将会是C1-0, C2-0, C3-0。然后将 partitions 的个数除于消费者线程的总数来决定每个消费者线程消费几个分区。如果除不尽，那么前面几个消费者线程将会多消费一个分区。在我们的例子里面，我们有 10 个分区，3 个消费者线程， 10 / 3 = 3，而且除不尽，那么消费者线程 C1-0 将会多消费一个分区，所以最后分区分配的结果看起来是这样的：

C1-0 将消费 0, 1, 2, 3 分区

C2-0 将消费 4, 5, 6 分区

C3-0 将消费 7, 8, 9 分区

假如我们有 11 个分区，那么最后分区分配的结果看起来是这样的：

C1-0 将消费 0, 1, 2, 3 分区

C2-0 将消费 4, 5, 6, 7 分区

C3-0 将消费 8, 9, 10 分区

假如我们有 2 个主题(T1 和 T2)，分别有 10 个分区，那么最后分区分配的结果看起来是这样的：

C1-0 将消费 T1 主题的 0, 1, 2, 3 分区以及 T2 主题的 0, 1, 2, 3 分区

C2-0 将消费 T1 主题的 4, 5, 6 分区以及 T2 主题的 4, 5, 6 分区

C3-0 将消费 T1 主题的 7, 8, 9 分区以及 T2 主题的 7, 8, 9 分区

可以看出，C1-0 消费者线程比其他消费者线程多消费了 2 个分区，这就是 Range strategy 的一个很明显的弊端

RoundRobin strategy（轮询分区）

轮询分区策略是把所有 partition 和所有 consumer 线程都列出来，然后按照 hashcode 进行排序。最后通过轮询算法分配 partition 给消费线程。如果所有 consumer 实例的订阅是相同的，那么 partition 会均匀分布。

在我们的例子里面，假如按照 hashCode 排序完的 topicpartitions 组依次为 T1-5, T1-3, T1-0, T1-8, T1-2, T1-1, T1-4, T1-7, T1-6, T1-9，我们的消费者线程排序为 C1-0, C1-1, C2-0, C2-1，最后分区分配的结果为：

C1-0 将消费 T1-5, T1-2, T1-6 分区；

C1-1 将消费 T1-3, T1-1, T1-9 分区；

C2-0 将消费 T1-0, T1-4 分区；

C2-1 将消费 T1-8, T1-7 分区；

使用轮询分区策略必须满足两个条件

每个主题的消费者实例具有相同数量的流

每个消费者订阅的主题必须是相同的

什么时候会触发这个策略呢？

当出现以下几种情况时，kafka 会进行一次分区分配操作，

也就是 kafka consumer 的 rebalance

同一个 consumer group 内新增了消费者

消费者离开当前所属的 consumer group，比如主动停机或者宕机

topic 新增了分区（也就是分区数量发生了变化）

kafka consuemr 的 rebalance 机制规定了一个 consumergroup 下的所有 consumer 如何达成一致来分配订阅 topic的每个分区。而具体如何执行分区策略，就是前面提到过的两种内置的分区策略。而 kafka 对于分配策略这块，提供了可插拔的实现方式， 也就是说，除了这两种之外，我们还可以创建自己的分配机制。

什么时候会触发这个策略呢？

当出现以下几种情况时，kafka 会进行一次分区分配操作，也就是 kafka consumer 的 rebalance

同一个 consumer group 内新增了消费者

消费者离开当前所属的 consumer group，比如主动停机或者宕机

topic 新增了分区（也就是分区数量发生了变化）kafka consuemr 的 rebalance 机制规定了一个 consumergroup 下的所有 consumer 如何达成一致来分配订阅 topic的每个分区。而具体如何执行分区策略，就是前面提到过的两种内置的分区策略。而 kafka 对于分配策略这块，提供了可插拔的实现方式， 也就是说，除了这两种之外，我们还可以创建自己的分配机制。

谁来执行 Rebalance 以及管理 consumer 的 group 呢？

Kafka 提供了一个角色： coordinator 来执行对于 consumer group 的管理，Kafka 提供了一个角色：coordinator 来执行对于 consumer group 的管理，当 consumer group 的第一个 consumer 启动的时候，它会去和 kafka server 确定谁是它们组的 coordinator。之后该 group 内的所有成员都会和该 coordinator 进行协调通信

如何确定 coordinator

consumer group 如何确定自己的 coordinator 是谁呢, 消费 者 向 kafka 集 群 中 的 任 一 个 broker 发 送 一 个GroupCoordinatorRequest 请求，服务端会返回一个负载最 小 的 broker 节 点 的 id ， 并 将 该 broker 设置为coordinator

JoinGroup 的过程

在 rebalance 之前，需要保证 coordinator 是已经确定好了的，整个 rebalance 的过程分为两个步骤，Join 和 Syncjoin: 表示加入到 consumer group 中，在这一步中，所有的成员都会向 coordinator 发送 joinGroup 的请求。一旦所有成员都发送了 joinGroup 请求，那么 coordinator 会选择一个 consumer 担任 leader 角色，并把组成员信息和订阅信息发送消费者

protocol_metadata: 序列化后的消费者的订阅信息

leader_id： 消费组中的消费者，coordinator 会选择一个座位 leader，对应的就是 member_id

member_metadata 对应消费者的订阅信息

members：consumer group 中全部的消费者的订阅信息

generation_id： 年代信息，类似于之前讲解 zookeeper 的时候的 epoch 是一样的，对于每一轮 rebalance ，

generation_id 都会递增。主要用来保护 consumer group。隔离无效的 offset 提交。也就是上一轮的 consumer 成员无法提交 offset 到新的 consumer group 中。

Synchronizing Group State 阶段

完成分区分配之后，就进入了 Synchronizing Group State阶段，主要逻辑是向 GroupCoordinator 发送SyncGroupRequest 请求，并且处理 SyncGroupResponse响应，简单来说，就是 leader 将消费者对应的 partition 分配方案同步给 consumer group 中的所有 consumer

每个消费者都会向 coordinator 发送 syncgroup 请求，不过只有 leader 节点会发送分配方案，其他消费者只是打打酱油而已。当 leader 把方案发给 coordinator 以后，coordinator 会把结果设置到 SyncGroupResponse 中。这样所有成员都知道自己应该消费哪个分区。

➢ consumer group 的分区分配方案是在客户端执行的！Kafka 将这个权利下放给客户端主要是因为这样做可以有更好的灵活性

5.1.3 如何保存消费端的消费位置

什么是 offset

前面在讲partition 的时候，提到过 offset， 每个 topic可以划分多个分区（每个 Topic 至少有一个分区），同一topic 下的不同分区包含的消息是不同的。每个消息在被添加到分区时，都会被分配一个 offset（称之为偏移量），它是消息在此分区中的唯一编号， kafka 通过 offset 保证消息在分区内的顺序， offset 的顺序不跨分区，即 kafka 只保证在同一个分区内的消息是有序的； 对于应用层的消费来说，每次消费一个消息并且提交以后，会保存当前消费到的最近的一个 offset。那么 offset 保存在哪里？

offset 在哪里维护？

在 kafka 中，提供了一个__consumer_offsets_的一个topic ，把 offset 信 息 写 入 到 这 个 topic 中。

_consumer_offsets——按保存了每个 consumer group某一时刻提交的 offset 信息。 consumer_offsets 默认有50 个分区。

5.1.4 消息的存储原理

消息的保存路径

消息发送端发送消息到 broker 上以后，消息是如何持久化的呢？那么接下来去分析下消息的存储

首先我们需要了解的是， kafka 是使用日志文件的方式来保存生产者和发送者的消息，每条消息都有一个 offset 值来表示它在分区中的偏移量。 Kafka 中存储的一般都是海量的消息数据，为了避免日志文件过大，Log 并不是直接对应在一个磁盘上的日志文件，而是对应磁盘上的一个目录，这个目录的明明规则是<topic_name>_<partition_id>比如创建一个名为 firstTopic 的 topic，其中有 3 个 partition，那么在 kafka 的数据目录（/tmp/kafka-log）中就有 3 个目录，firstTopic-0~3

多个分区在集群中的分配

如果我们对于一个 topic，在集群中创建多个 partition，那么 partition 是如何分布的呢？

1.将所有 N Broker 和待分配的 i 个 Partition 排序

2.将第 i 个 Partition 分配到第(i mod n)个 Broker 上

了解到这里的时候，大家再结合前面讲的消息分发策略，就应该能明白消息发送到 broker 上，消息会保存到哪个分区中，并且消费端应该消费哪些分区的数据了。

消息写入的性能

我们现在大部分企业仍然用的是机械结构的磁盘，如果把消息以随机的方式写入到磁盘，那么磁盘首先要做的就是寻址，也就是定位到数据所在的物理地址，在磁盘上就要找到对应的柱面、磁头以及对应的扇区；这个过程相对内存来说会消耗大量时间，为了规避随机读写带来的时间消耗， kafka 采用顺序写的方式存储数据。即使是这样，但是频繁的 I/O 操作仍然会造成磁盘的性能瓶颈，所以 kafka还有一个性能策略

5.1.5 页缓存

顺序写入是Kafka高吞吐量的一个原因，当然即使采用的是磁盘的顺序写入，那么也是没有办法和内存相比的。因为为了再一次提高Kakfa的吞吐量，Kafka采用了Memory Mapped Files

(后面简称mmap)也被翻译成内存映射文件 ，它的工作原理是直接利用操作系统的page cache 来实现文件到物理内存的直接映射，完成映射之后你对物理内存的操作会被同步到硬盘上（操作系统在适当的时候）。

操作系统本身有一层缓存，叫做page cache，是在内存里的缓存，我们也可以称之为os cache，意思就是操作系统自己管理的缓存。你在写入磁盘文件的时候，可以直接写入这个os cache里，也就是仅仅写入内存中，接下来由操作系统自己决定什么时候把os cache里的数据真的刷入磁

盘文件中（每5秒检查一次是否需要将页缓存数据同步到磁盘文件）。仅仅这一个步骤，就可以将磁盘文件写性能提升很多了，因为其实这里相当于是在写内存，不是在写磁盘.

5.1.6 零拷贝

消息从发送到落地保存，broker 维护的消息日志本身就是文件目录，每个文件都是二进制保存，生产者和消费者使用相同的格式来处理。在消费者获取消息时，服务器先从硬盘读取数据到内存，然后把内存中的数据原封不动的通过 socket 发送给消费者。虽然这个操作描述起来很简单，但实际上经历了很多步骤。

操作系统将数据从磁盘读入到内核空间的页缓存

▪ 应用程序将数据从内核空间读入到用户空间缓存中

▪ 应用程序将数据写回到内核空间到 socket 缓存中

▪ 操作系统将数据从 socket 缓冲区复制到网卡缓冲区，以便将数据经网络发出

这个过程涉及到 4 次上下文切换以及 4 次数据复制，并且有两次复制操作是由 CPU 完成。但是这个过程中，数据完全没有进行变化，仅仅是从磁盘复制到网卡缓冲区。

通过“零拷贝”技术，可以去掉这些没必要的数据复制操作，同时也会减少上下文切换次数。现代的 unix 操作系统提供一个优化的代码路径，用于将数据从页缓存传输到 socket；在 Linux 中，是通过 sendfile 系统调用来完成的。Java 提供了访问这个系统调用的方法： FileChannel.transferTo API

使用 sendfile，只需要一次拷贝就行，允许操作系统将数据直接从页缓存发送到网络上。所以在这个优化的路径中，只有最后一步将数据拷贝到网卡缓存中是需要的