系统间的耦合高怎么办，我们如何不让一个服务过于庞大，一个好的方式就是依据具体的功能模块拆分服务，降低服务的耦合度，服务间的交互可以通过消息传递数据来实现，除此之外Kafka非常适合在线日志收集等高吞吐场景，kafka有更好的吞吐量，内置分区，副本和故障转移，这有利于处理大规模的消息，所以kafka被各大公司广泛运用于消息队列的构建：
- 消息队列模型-生产者消费者模型
- Kafka基本概念和架构模型
- Kafka工作流程和文件存储机制
- 生产者策略：分区策略、ACK机制、故障转移机制、Kafka可靠高效原因
- 消费者策略：消费方式、分区分配策略、offset的维护
- Zookeeper管理
- Kafka框架搭建实战
适合人群：不了解Kafka的新手，对Kafka的实现机制感兴趣的技术人员
本文的全部内容来自我个人在Kafka学习过程中整理的博客，是该博客专栏的精华部分。在书写过程中过滤了流程性的上下文，例如部署环境、配置文件等，而致力于向读者讲述其中的核心部分，如果读者有意对过程性内容深入探究，可以移步MaoLinTian的Blog，在这篇索引目录里找到答案分布式技术相关专栏索引
消息队列模型-生产者消费者模型
在正式介绍Kafka之前，先来了解下什么是生产者消费者模式，以及什么是消息队列，再由消息队列聊到Kafka。
生产者消费者模型具体来讲，就是在一个系统中，存在生产者和消费者两种角色，他们通过内存缓冲区进行通信，生产者生产消费者需要的资料，消费者把资料做成产品。可以理解为生产者不停的生产馒头，然后把馒头扔到筐里，消费者不停的从筐里拿出馒头吃。具体到消息系统可以这么理解：
- 生产者：消息的制造者，生产馒头的一方
- 管道容器：消息的传递者，也就是放馒头的筐
- 消费者：消息的消费者，也就是吃馒头的一方
其中最重要的环节就是这个管道容器，其实就是一个消息队列
消息队列的应用场景
消息队列具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能，成为异步RPC的主要手段之一。那么为什么要使用队列呢？我这么理解：
- 1，在分布式高并发场景下，如何保证系统性能不受影响，请求不会超时中断并且能让用户无感知（削峰、减少响应所需时间）,一般网站响应时间超过200ms就难以忍受了，高并发同步向数据库内写数据，数据库会扛不住压力的，而且响应速度也会明显减慢。
- 2，不同系统之间的异步通信如何实现（ 降低系统耦合性）,一个系统必然有很多模块组成，不同模块各自有实现，各自有自己的迭代，如果耦合性低，不存在强依赖，那最好了，微服务架构的思想也类似此，降低耦合性。
第一个场景下，例如有个业务：下班打车，企业滴滴，晚上8点半上地这边爆满，各种打车订单同时涌入系统，系统要爆，加上消息队列后，订单进入消息队列，然后反馈给用户订单在处理，然后按照入队顺序一个个处理该消息，直到消费到该消息订单后才将订单入库，当然这个时候才能反馈给用户说打车成功。这样有效的抵抗了峰值时间的冲击。
第二个场景下，系统的不同模块之间各自有各自的实现。模块之间不存在直接调用，那么新增模块或者修改模块就对其他模块影响较小，所以如果有个消息系统在中间作为中转，消息发送者将消息发送至分布式消息队列即结束对消息的处理，消息接受者从分布式消息队列获取该消息后进行后续处理，并不需要知道该消息从何而来。对新增业务，只要对该类消息感兴趣，即可订阅该消息，对原有系统和业务没有任何影响，从而实现网站业务的可扩展性设计。
消息队列模型
消息队列目前有两种协议：JMS和AMQP，JMS轻量不能跨平台，AMQP性能强但是复杂，感觉就是美图秀秀和PS的区别，但是大家用的多的还是美图秀秀，那么就以JMS为主吧，咱就好好用轮子吧。
1 点到点（P2P）模型
使用队列（Queue）作为消息通信载体：满足生产者与消费者模式，一条消息只能被一个消费者使用，未被消费的消息在队列中保留直到被消费或超时。
假如我们存在这样⼀种情况：我们需要将⽣产者产⽣的消息分发给多个消费者，并且每个消费者都能接收到完全的消息内容。这种情况，队列模型就不好解决了
2 发布/订阅（Pub/Sub）模型
发布订阅模型（Pub/Sub） 使用主题（Topic）作为消息通信载体，类似于广播模式,发布者发布一条消息，该消息通过主题传递给所有的订阅者，一条消息可以被多个消费者使用。
在一条消息广播之后才订阅的用户则是收不到该条消息的。在发布 - 订阅模型中，如果只有⼀个订阅者，那它和队列模型就基本是⼀样的了。所以说，发布 - 订阅模型在功能层⾯上是可以兼容点到点（P2P）模型的
消息队列问题
系统里平白无故加入一个消息队列服务器当然会影响系统的整体性能，增加系统的复杂性，还需要担心异步消息发送的一致性问，消息服务器的稳定性等问题：
- 系统可用性降低： 系统可用性在某种程度上降低，在加入MQ之前，不用考虑消息丢失或者说MQ挂掉等等的情况，但是，引入MQ之后就需要去考虑了
- 系统复杂性提高： 加入MQ之后，需要保证消息没有被重复消费、处理消息丢失的情况、保证消息传递的顺序性，消息服务器有没有宕机等问题
- 一致性问题： 消息队列可以实现异步，异步确实可以提高系统响应速度。但是，万一消息的真正消费者并没有正确消费消息就会导致数据不一致的情况了。
其实这些问题就是作为一个消息队列中间件所要面对的挑战，这个消息中间件该如何设计才能解决消息框架可能遇到的一系列问题：故障转移恢复、数据一致性保证、数据可靠性保证
Kafka基本概念和架构模型
了解了生产者-消费者模型以及消息队列模型后，我们知道如果系统需要解耦、削峰，一定要使用消息队列，但是消息队列存在诸多问题，所以需要一个功能完备的中间件来解决这些问题。
Kafka 是⼀个分布式流式处理平台，与大多数消息系统比较，kafka有更好的吞吐量，内置分区，副本和故障转移。kafka流平台具有三个关键功能：
- 消息队列【传递消息】：发布和订阅消息流，这个功能类似于消息队列，这也是 Kafka 也被归类为消息队列的原因。
- 容错的持久方式存储记录消息流【存储消息】： Kafka 会把消息持久化到磁盘，有效避免了消息丢失的⻛险·。
- 流式处理平台【处理消息】： 在消息发布的时候进⾏处理，Kafka 提供了⼀个完整的流式处理类库。
所以Kafka 主要有两⼤应⽤场景：首先基于传递和存储消息的有效和可靠性，kafka可以建⽴实时流数据管道，以可靠地在系统或应⽤程序之间获取数据。其次基于处理消息数据流的完备能力，kafka可以构建实时的流数据处理程序来转换或处理数据流。
Kafka架构模型
看了那么多的Kafka的优点，那么相信各位对Kafka的实现架构有了一些好奇，下图为一个Kafka的基本架构图，接下来我会对其中的概念分点详述并举例说明
1 Producer（生产者）：负责发布消息到Kafka broker，就是向kafka broker发消息的客户端。
2 主题和分区：一个主题包含一个或多个Partition
- Topics（主题）: 属于特定类别的消息流称为主题。 数据存储在主题中。主题被拆分成分区。 对于每个主题，Kafka保存一个分区的数据。 每个这样的分区包含不可变有序序列的消息。 分区被实现为具有相等大小的一组分段文件。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处），也可以理解为一个队列，通过对消息指定主题可以将消息分类，消费者可以只关注自己需要的Topic中的消息。
- Partition（分区），Parition是物理上的概念，每个Topic包含一个或多个Partition。每个Partition包含N个副本。为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列， partition中的每条消息都会被分配一个有序的id（offset 分区偏移量）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序，
- Leader（领导者）【针对某个分区的Leader副本】： Leader 负责指定分区的所有读取和写入的操作。 每个分区都有一个服务器充当Leader。
- Follower（追随者）【针对某个分区的Follower副本】：跟随领导者指令的节点被称为Follower。 如果领导失败，一个追随者将自动成为新的领导者。 跟随者作为正常消费者，拉取消息并更新其自己的数据存储。 follower从不用来读取或写入数据， 它们用于防止数据丢失。
3 Kafka集群和Kafka服务器：一个Kafka集群包含多个Kafka服务器
- Kafka Cluster（Kafka集群）：Kafka有多个服务器被称为Kafka集群。 可以扩展Kafka集群，无需停机。 这些集群用于管理消息数据的持久性和复制。
- Broker（Kafka服务器）：一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。
4 Consumer(消费者)和 Consumer Group（消费者集群）：一个消费者集群包含多个消费者
- Consumer Group（消费者集群）：consumer group是kafka提供的可扩展且具有容错性的消费者机制。组内可以有多个消费者或消费者实例，它们共享一个公共的group ID。组内的所有消费者协调在一起来共享订阅主题的所有分区。
- Consumer(消费者)：消息消费者，向Kafka broker读取消息的客户端。
综合而言我理解就是，Kafka集群和Kafka服务器属于物理机器上的概念，而主题和分区属于发出去的消息的分类，一个纵向，一个横向，一个broker上可以有很多主题的分区，一个主题也可以在很多broker上放置分区，是多对多的关系。由于这部分比较难理解，这里我举个例子：
Kafka概念举例
我在有三台机器【broker的id分别为101、102、103】的Kafka集群上创建了一个主题为tml-kafka的集群，指定副本数为3，查看详情时第一个行显示所有partitions的一个总结，以下每一行给出一个partition中的信息，如果我们只有一个partition，则只显示一行。
- leader 和follwer都是针对某个分区的概念，例如对于分区0，其leader在0也就是broker为101的机器，而对于分区1和分区2其leader在2也就是broker为103的机器上
- Replicas 显示给定partiton所有副本所存储节点的节点列表，不管该节点是否是leader或者是否存活，这里就是我们的三台机器、0、1、2，分别对应101、102、103。
- Isr 副本都已同步的的节点集合，这个集合中的所有节点都是存活状态，并且跟leader同步,这里也是我们的三台机器、0、1、2，分别对应101、102、103.说明我们三台机器都在集群里没有脱机，ISR的概念在后续会提到。
Kafka工作流程和文件存储机制
了解了Kafka的基本架构和示例后我们来了解下Kafka到底是怎么工作的，以及消息是如何在Kafka持久化存储的。
Kafka的工作流程
通过基础概念的学习可以知道kafka的消息分为Topic，而Topic从逻辑上又可以划分为Partion，关于Topic&Partion需要注意以下几点：
- 一个 Topic可以认为是一类消息，每个 topic 将被分成多个 Partion，每个 Partion在存储层面是 append log 文件。Kafka 机制中，producer push 来的消息是追加（append）到 Partion中的，这是一种顺序写磁盘的机制，效率远高于随机写内存
- 任何发布到partition 的消息都会被追加到log文件的尾部，每条消息在文件中的位置称为 offset（偏移量），offset 为一个 long 型的数字，它唯一标记一条消息。Kafka只保证Partion内的消息有序，不能保证全局Topic的消息有序
- 消费者组中的每个消费者，都会实时记录自己消费到了哪个 offset，以便出错恢复时，从上次的位置继续消费
下图为一个消息写入过程，Producer向同一主题的不同分区写消息，也即不停的在各个append log文件后顺序追加消息，每追加一个append log文件偏移量加一，只有单append log文件中有序
整体消息的生产传递和消费的的流程如下图所示，注意这里偏移量数字从consumer消费的视角来看，无论是生产者还是消费者对消息的处理都是偏移量从小到大的：
消息由生产者生产，并发往Kafka集群，Kafka集群获取到消息后再发往消费者，以下是Pub-Sub消息的逐步工作流程 ：
- Producer生产者定期向Kafka集群发送消息，在发送消息之前，会对消息进行分类，即Topic, Kafka集群存储该Topic配置的分区中的所有消息。 它确保消息在分区之间平等共享。 如果生产者发送两个消息并且有两个分区，Kafka将在第一分区中存储一个消息，在第二分区中存储第二消息。
- Kafk接收生产者消息并转发给消费者，消费者订阅特定主题，一旦消费者订阅Topic，Kafka将向消费者提供Topic下分区的当前offset，并且还将偏移保存在Zookeeper系统中。消费者通过与kafka集群建立长连接的方式，不断地从集群中拉取消息，然后可以对这些消息进行处理，一旦Kafka收到来自生产者的消息，它将这些消息转发给消费者。
- 消费者将收到消息进行处理,一旦消息被处理，消费者将向Kafka代理发送确认。消费者需要实时的记录自己消费到了哪个offset，便于后续发生故障恢复后继续消费。Kafka 0.9版本之前，consumer默认将offset保存在Zookeeper中，从0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中 一旦Kafka收到确认，它将offset更改为新值，并在Zookeeper中更新它。
以上流程将重复，直到消费者停止请求。消费者可以随时回退/跳到所需的主题偏移量，并阅读所有后续消息
Kafka文件存储机制
我们上文提到每个partion是一个append log 文件，虽然我们已经把Topic物理上划分为多个Partion用来负载均衡，但即使是对一个Partition 而言，如果消息量过大的话也会有堵塞的风险，所以我们需要定期清理消息。
清理消息时如果只有一个Partion，那么就得全盘清除，这将对消息文件的维护以及已消费的消息的清理带来严重的影响。所以我们需要在物理上进一步细分Partition
- Kafka以 segment 为单位将 partition 进一步细分，每个 partition（目录）相当于一个巨型文件被平均分配到多个大小相等的 segment（段）数据文件中（每个 segment 文件中消息数量不一定相等）
这种特性也方便 old segment 的删除，即方便已被消费的消息的清理，提高磁盘的利用率，每个 partition 只需要支持顺序读写就行。
Partition&Segment
接下来我们发送一条消息，看看在物理存储上是什么样的，在机器上我们可以看到，一组index和log，这就是一个segment的内容：
segment 文件由两部分组成，分别为 “.index” 文件和 “.log” 文件，分别表示为 segment 索引文件和数据文件。这两个文件的命令规则为：partition 全局的第一个 segment 从 0 开始，后续每个 segment 文件名为上一个 segment 文件最后一条消息的 offset 值，数值大小为 64 位，20 位数字字符长度，没有数字用 0 填充，我这里只有一条数据，所以是从0开始的，整体的存储架构如下：
Segment存储结构
通过以上对Segment落盘文件的了解，我们基本搞清楚了Segment的结构，当然我这里是单segment，看不出来。这里从网上找了一个大量文件的示例：
//第一段segment，起始位置为0 00000000000000000000.index 00000000000000000000.log //第二段segment，起始位置为170410 00000000000000170410.index 00000000000000170410.log //第三段segment，起始位置为239430 00000000000000239430.index 00000000000000239430.log
以上面的 segment 文件为例，展示出 segment：00000000000000170410 的 “.index” 文件和 “.log” 文件的对应的关系，如下图：
如上图，“.index” 索引文件存储大量的元数据，“.log” 数据文件存储大量的消息，索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。其中以 “.index” 索引文件中的元数据 [3, 348] 为例，在 “.log” 数据文件表示第 3 个消息，即在全局 partition 中表示 170410+3=170413 个消息，该消息的segementoffset为3，全局offset为170413，物理偏移地址为 348（注意此物理偏移地址不是消息数量的offset，而是消息的内存存储偏移量 ）
快速定位partion中消息
既然消息在Partion中被分为了一段段的segment，那么我们如何快速定位消息的位置，来精准的对消息进行操作呢？以上图为例，读取 offset=170418 的消息：
- 首先查找 segment 文件，其中 00000000000000000000.index 为最开始的文件，第二个文件为 00000000000000170410.index（起始偏移为 170410+1=170411），而第三个文件为 00000000000000239430.index（起始偏移为 239430+1=239431），所以这个 offset=170418 就落到了第二个文件之中。其它后续文件依次类推，以其偏移量命名并排列这些文件，然后根据二分查找法就可以快速定位到具体文件位置。
- 其次根据 00000000000000170410.index 文件中的170418 -170410=8，得出是该segment 中的第8个消息，再次依据二分查找法定位到该索引，得到segment内[消息偏移量，物理偏移量]坐标 [8,1325] 定位到 00000000000000170410.log 文件中的 1325 的位置进行读取。
- 找到1325位置后，顺序读取消息即可，确定读完本条消息【本条消息读到哪里结束】由消息的物理结构解决，消息都具有固定的物理结构，包括：offset（8 Bytes）、消息体的大小（4 Bytes）、crc32（4 Bytes）、magic（1 Byte）、attributes（1 Byte）、key length（4 Bytes）、key（K Bytes）、payload（N Bytes）等等字段，可以确定一条消息的大小，即读取到哪里截止。
以上就是定位消息的详细方法，通过索引的方式，可以在kafka顺序写磁盘的基础上仍然能快速的找到对应的消息。
生产者策略：分区策略、数据可靠性保证、故障转移机制
前面提到消息系统的三大问题： 系统可用性降低，系统复杂性提高，数据一致性问题，为了解决这些问题，Kafka支持一些机制，例如：
- 使用分区策略来提高系统可用性和进行负载均衡【高可扩展】
- 使用ACK应答机制来保障数据的可靠性【副本同步策略、ISR、Exactly Once语义】保证的一系列策略来解决系统复杂性问题，例如保证消息的不重不漏【高并发】
- 使用故障转移机制【HW&LEO概念、故障同步机制、Leader选举】来保证消息的数据一致性，防止意外宕机丢数据导致不一致【高可用】
接下来就这三个机制以及其附加的策略来进行详细的讨论。
