中标麒麟的系统介绍

桌面操作系统

中标麒麟桌面操作系统是一款面向桌面应用的图形化桌面操作系统，针对X86及龙芯、申威、众志、飞腾等国产CPU平台进行自主开发，率先实现了对X86及国产CPU平台的支持， 提供性能最优的操作系统产品。通过进一步对硬件外设的适配支持、对桌面应用的移植优化和对应用场景解决方案的构建，完全满足项目支撑，应用开发和系统定制的需求。

该系统除了具备基本功能外，还可以根据客户的具体要求，针对特定软硬件环境，提供定制化解决方案，实现性能优化和个性化功能定制。

中标麒麟桌面操作系统是国家重大专项的核心组成部分，是民用、军用“核高基”项目桌面操作系统项目的重要研究成果，该系统成功通过了多个国家权威部门的测评，为实现操作 系统领域“自主可控” 的战略目标做出了重大贡献。在国产操作系统领域市场占有率稳居第一。

中标麒麟桌面操作系统针对X86及龙芯、申威、众志等国产CPU平台，完成了硬件适配、软件移植、功能定制和性能优化，可以运行在台式机、笔记本、一体机、 车载机等不同产品形态之上，支撑着国防、政府、企业、电力和金融等各领域的应用。

安全操作系统

为满足政府、国防、电力、金融、证券等领域，以及企业电子商务和互联网应用对操作系统平台的安全需求，中标软件有限公司研发的安全可控、高安全等级的操作系统平台产品——中标麒麟安全操作系统软件V50；为用户提供全方位的操作系统和应用安全保护，防止关键数据被篡改被窃取，系统免受攻击，保障关键应用安全、可控和稳定的对外提供服务。

安全邮件服务器

为了解决通用邮件系统安全性和保密性问题，中标软件有限公司面向政府和中小企业研制推出的安全增强电子邮件服务器软件。

中标麒麟安全邮件服务器软件基于高安全等级的中标麒麟安全操作系统平台并与之集成一体化，构建“系统安全+邮件安全+管理安全+防攻击”结合的全方位立体化安全体系，确保用户能够防御来自内部人员及外部攻击的邮件窃密。

安全云操作系统

中标麒麟安全云操作系统目包含三个组件：

(1) 中标麒麟安全云管理平台（NeoKylin Security Cloud Management Platform，简称NeoKylin sCloud MP

(2) 中标麒麟安全云虚拟化服务器（NeoKylin Security Cloud Virtualization Server，简称NeoKylin sCloud VS

(3) 中标麒麟安全云虚拟桌面套件（NeoKylin Security Cloud Virtual Desktop Suite，简称NeoKylin sCloud VDS

中标麒麟安全云管理平台（NeoKylin sCloud MP）

中标麒麟安全云管理平台基于J2EE体系架构构建，借助web20的优势，提供云计算基础架构各种资源（包括CPU、内存、网络、映像、虚拟机等）的抽象聚合、管理、调度、监控、 统计等功能，具备用户管理、组管理、权限控制、日志审计等安全管理手段，是中标麒麟安全云操作系统的核心管理组件。

中标麒麟安全云虚拟化服务器（NeoKylin sCloud VS）

中标麒麟安全云虚拟化服务器利用x86硬件虚拟化技术，提供虚拟机监控器，能够对虚拟机运行状态监控、管理并实施负载均衡， 其硬件兼容性好、功能稳定、易于线性扩展，是中标麒麟安全云操作系统的核心服务组件。

中标麒麟安全云虚拟桌面套件（NeoKylin sCloud VDS）

中标麒麟安全云虚拟桌面套件基于开源技术研发自主可控的虚拟桌面协议，并结合中标麒麟安全云管理平台，为用户提供了一套界面简洁易用、 认证机制安全可靠的虚拟桌面解决方案，是中标麒麟安全云操作系统的重要应用组件。

高级服务器操作系统

中标麒麟高级服务器操作系统是中标软件有限公司依照CMMi5标准研发、发行的国产Linux操作系统；针对关键业务及数据负载而构建的高可靠、易管理、一架式Linux服务器操作系统。

中标麒麟高级服务器操作系统提供中文化的操作系统环境和常用图形管理工具；支持多种安装方式，提供了完善的文件系统支持，系统服务，网络服务；集成了丰富易用的编译器和支持众多的开发语言；全面兼容国内外的软硬件厂商；同时在安全上进行了加强，保障关键应用安全、可控、稳定的对外提供服务。

基于中标麒麟高级服务器操作系统用户可以轻松构建大型数据中心、高可用集群和负载均衡集群，虚拟化应用服务、分布式文件系统等，同时可以方便的进行集中监控和管理。

经过多年的产品研发积累和市场拓展，中标麒麟服务器操作系统已经成长为国内Linux服务器操作系统的第一品牌。

通用服务器操作系统

中标麒麟通用服务器操作系统用于部署和管理中小型企业级和部门级应用服务，为用户提供高性能处理能力和高可靠性。中标麒麟通用服务器操作系统提供全图形化的系统配置与管理工具，减少维护人员的管理难度和再培训成本；通过基于MLS多级安全的SELinux增强和基于应用的安全优化实现系统的安全增强；丰富的开源网络服务方便用户轻松构建企业级应用；完整的开发平台提供对主流开发工具、开发语言支持。借助中标麒麟通用服务器操作系统，将进一步降低企业IT应用成本，为用户提供更高质量、更为可靠的系统平台服务。

高可用集群软件

中标麒麟高可用集群软件是基于中标麒麟服务器操作系统开发的智能高可用软件产品，是国内首款支持国产龙芯CPU架构的高可用产品，通过应用中标麒麟高可用集群产品可以提升软硬件系统及应用运行的稳定性和可靠性，该产品经过多年的用户应用及市场验证，提供的抗错能力足以支持关键业务系统应用可靠性苛刻要求为政府、金融、电力、医疗、运输、制造业、等行业用户提供高效、至微的可靠服务。

中标麒麟高可用集群软件依托“系统可靠”——“数据可靠”——“应用可靠”实现对业务系统的“智能、多层次可靠保护”：

系统可靠：保护用户的操作系统及硬件设备，对故障操作系统及硬件设备进行智能、快速、便捷的恢复。

数据可靠：为用户共享数据提供一致性保护，当系统出现脑裂等极端故障的情况下，保证数据不被破坏。

应用可靠：为用户业务系统稳定、高效、持续的运行提供缜密的可靠保护。

要下载rar解压软件。麒麟系统能解压zip等格式，没有解压rar的功能，要用户下载rar解压软件来解压rar压缩文件。麒麟系统是由上海中标软件有限公司开发的一款基于linux内核的电脑操作系统，属于国产系统。

1 主服务器的自动监控和故障转移

MHA监控复制架构的主服务器，一旦检测到主服务器故障，就会自动进行故障转移。即使有些从服务器没有收到最新的relay log，MHA自动从最新的从服务器上识别差异的relay log并把这些日志应用到其他从服务器上，因此所有的从服务器保持一致性了。MHA通常在几秒内完成故障转移，9-12秒可以检测出主服务器故障，7-10秒内关闭故障的主服务器以避免脑裂，几秒中内应用差异的relay log到新的主服务器上，整个过程可以在10-30s内完成。还可以设置优先级指定其中的一台slave作为master的候选人。由于MHA在slaves之间修复一致性，因此可以将任何slave变成新的master，而不会发生一致性的问题，从而导致复制失败。

2 交互式主服务器故障转移

可以只使用MHA的故障转移，而不用于监控主服务器，当主服务器故障时，人工调用MHA来进行故障故障。

3 非交互式的主故障转移

不监控主服务器，但自动实现故障转移。这种特征适用于已经使用其他软件来监控主服务器状态，比如heartbeat来检测主服务器故障和虚拟IP地址接管，可以使用MHA来实现故障转移和slave服务器晋级为master服务器。

4 在线切换主从服务器

在许多情况下，需要将现有的主服务器迁移到另外一台服务器上。比如主服务器硬件故障，RAID控制卡需要重建，将主服务器移到性能更好的服务器上等等。维护主服务器引起性能下降，导致停机时间至少无法写入数据。另外，阻塞或杀掉当前运行的会话会导致主主之间数据不一致的问题发生。MHA提供快速切换和优雅的阻塞写入，这个切换过程只需要05-2s的时间，这段时间内数据是无法写入的。在很多情况下，05-2s的阻塞写入是可以接受的。因此切换主服务器不需要计划分配维护时间窗口(呵呵，不需要你在夜黑风高时通宵达旦完成切换主服务器的任务)。

ES支持集群模式，是一个分布式系统，其好处主要有两个∶

es集群由多个ES 实例组成。不同集群通过集群名字来区分，可通过 clustername 进行修改，默认为elasticsearch。每个ES实例本质上是一个 JVM 进程，且有自己的名字，通过 nodename 进行修改

ES集群相关的数据称为 cluster state ，主要记录如下信息∶节点信息，比如节点名称、连接地址等；索引信息，比如索引名称、配置等

可以修改 cluster state 的节点称为master节点，一个集群只能有一个 cluster state 存储在每个节点上，master维护最新版本并同步给其他节点

master节点是通过集群中所有节点选举产生的，可以被选举的节点称为 master-eligible 节点 ，相关配置如下: nodemaster: true

处理请求的节点即为coordinating节点，该节点为所有节点的默认角色，不能取消。路由请求到正确的节点处理，比如创建索引的请求到master节点

存储数据的节点即为data节点，默认节点都是data类型，相关配置如下∶ nodedata: true

谈及副本和分片两个概念之前，我们先说一下这两个概念存在的意义： 解决系统可用性和增大系统容量

我们想象这样一个场景，我们的数据只存放在一台ES服务器上，那么一旦这台ES出现宕机或者其他不可控因素影响的话，我们除了丧失了服务的可用性外，可能还存在着数据丢失的可能。同时，单机服务的存储容量也无法应对项目对大数据量的要求。

系统可用性可以分为 服务可用性 和 数据可用性

服务可用性 含义为：当前服务挂掉后，是否有其他服务器顶替当前节点提供服务支持。

数据可用性 含义为：当前服务挂掉后，存储在当前服务器上的数据，是否还可以对外提供访问和修改的服务。

副本可以理解为是某个数据的复制体，副本和源数据内容一致。副本的存在可以有效地满足系统可用性的需求，比如说，我们可以在原有节点的基础上复制一个和源节点一模一样的节点，这样一旦原有节点挂掉了，另外一个节点也还是可以替代源节点提供服务，而且复制出来的节点拥有和源节点一样的数据，这样也保障了数据可用性。

我们在上一小节讲到可以使用副本来解决系统可用性的问题，但是这里存在一个问题，不管存在多少个副本（节点），都无法增大源节点的存储空间。在这个问题上，ES引入了Shard分片这个概念来解决问题。

看完分片的特点后可能还有人不太清楚到底什么是分片，其实分片是n/1个源节点数据。比如说原ES集群中只有一个主节点，所有的索引数据都存储在这个节点上。现在我们将某个索引数据分成3份，分别存放在3个ES节点上，那么每台ES服务器上就各自有1个分片shard。该索引的所有节点Shard分片的集合，就是索引的全部数据。

下面我们来演示一下：

为了更好的了解ES的分片机制，大家不妨在上面的案例上进一步思考两个问题：

答案是不能。原因是我们创建索引时定义的分片数量只有3个，且都已经落在了3个节点上。所以即使再增加多一个节点，也不会有对应的Shard分片可以落在新的节点上，并不能扩大 test_shard_index 的数据容量。

答案是不能。因为新增的副本也是分布在这3个节点上，还是利用了同样的资源。如果要增加吞吐量，还需要新增节点。

通过上面两个问题，相信大家已经可以认识到分片的重要性，分片数过小，会导致后续无法通过增加节点实现水平扩容；（副本）分片数过大会导致一个节点上分布过多分片，造成资源浪费，同时会影响查询性能

集群健康状况，包括以下三种: green健康状态，指所有主副分片都正常分配； yellow指所有主分片都正常分配，但是有副本分片未正常分配； red表示有主分片未分配

我们可以通过这个api查看集群的状态信息： GET _cluster/health

我们也可以通过cerebro或者head插件来直接获取当前集群的状态

需要注意的是，即使当前集群的状态为 red ，也并不代表当前的ES丧失了提供服务的能力。只是说未被分配主分片的索引无法正常存储和操作而已。

这里故障转移的意思是，当ES集群出现某个或者多个节点宕机的情况，ES实现服务可用性的应对策略。

这里我们新建一个分片为3，副本为1的索引，分片分别分布在三个节点，此时集群为 green

当master节点所在机器宕机导致服务终止，此时集群会如何处理呢

我们可以看到，从node1主节点宕机到ES恢复集群可用性的过程中，ES有着自己的故障转移机制，保障了集群的高可用性。我们也可以在自己的本地上去进行试验，建好索引后，kill掉主节点，观察集群状态就行。

同时，此时就算node2宕机了，那么node3也能够很快的恢复服务的提供能力。

我们知道，我们创建的文档最终会存储在分片上，那么在分布式集群的基础上，ES集群是怎么判断当前该文档最终应该落在哪一个分片上呢？

很显然，我们需要一个可以实现文档均匀分布到各个分片上的映射算法，那么常见的随机算法和round-robin（轮询）算法可以满足需要吗？答案是不可以，这两个算法虽然可以实现文档均匀分布分片的存储需要，但是当我们通过 DocumentId 查询文档时，ES并不能知道这个文档ID到底存储在了哪个节点的分片上，所以只能够从所有分片上检索，时间长。如果我们为这个问题建立一个文档和分片映射关系的表，虽然确实可以快速定位到文档对应的存储分片，但是当文档的数据量很大的时候，那么检索的效率也会随之变低。

对于上面这个问题，ES提供的解决方法是 建立文档到分片的映射算法

es 通过如下的公式计算文档对应的分片：

hash算法 保证可以将数据均匀地分散在分片中

routing 是一个关键参数，默认是文档id，也可以自行指定

number_of_primary_shards 是主分片数

我们可以看到，该算法与主分片数相关， 这也是分片数一旦确定后便不能更改的原因

我们已经知道了ES是如何将文档映射到分片上去了，下面我们就来详细讲解一下文档创建、读取的流程。

脑裂问题，英文为 split-brain ，是分布式系统中的经典网络问题，如下图所示:

3个节点组成的集群，突然node1的网络和其他两个节点中断

解决方案为 仅在可选举master-eligible节点数大于等于quorum时才可以进行master选举

在讲文档搜索实时性之前，先讲一下倒排索引的不可变更特性。由于倒排索引一旦生成，不可变更的特定，使得其有着以下3点好处：

下面，将针对Lucene实现文档实时性搜索的几个动作进行讲解，分析其是如何在新增文档后实现ES的搜索实时性的。

我们从上面的描述中知道，当我们新增了一个文档后会新增一个倒排索引文件 segment ，但是 segment 写入磁盘的时间依然比较耗时（难以实现实时性），所以ES借助文件系统缓存的特性， 先将 segment 在缓存中创建并开放查询来进一步提升实时性 ，该过程在es中被称为refresh。

在refresh之前文档会先存储在一个buffer中，refresh时将 buffer中的所有文档清空并生成 segment

es默认每1秒执行一次refresh，因此文档的实时性被提高到1秒 ，这也是es被称为近实时(Near Real Time)的原因

reflush虽然通过 将文档存放在缓存中 的方式实现了秒级别的实时性，但是如果在内存中的segment还没有写入磁盘前发生了宕机，那么其中的文档就无法恢复了，如何解决这个问题呢

ES 引入 translog 机制。写入文档到 buffer 时，同时将该操作写入 translog 中。

translog文件会即时写入磁盘(fsync)，在ES 6x中，默认每个请求都会落盘，我们也可以修改为每5秒写一次，这样风险便是丢失5秒内的数据，相关配置为indextranslog。同时ES每次启动时会检查translog 文件，并从中恢复数据。

flush 负责将内存中的segment写入磁盘，主要做如下的工作:

Reflush和Flush执行的时机

ES的做法是 首先删除文档，然后再创建新文档

我们上面提到，新增文档是通过新建segment来解决，删除文档是通过维护del文件来进行的，假如现在我们设置的 reflush 时间间隔为1秒，那么一小时单个ES索引就会生成3600个segment，一天下来乃至一个月下来会产生的segment文件数量更是不可想象。为了解决Segment过多可能引起的性能下降问题，ES中采用了Segment Merging（即segment合并）的方法来减少segment的数量。

执行合并操作的方式有两种，一种是ES定时在后台进行 Segment Merging 操作，还有一种是我们手动执行 force_merge_api 命令来实现合并操作。

在学习一样技术之前，咱们需要先想一下，为什么需要学这一门技术？

许多分布式系统都是基于ZK作为底层核心组件对外提供服务，比如Kafka中，将Broker注册到ZK中，此时ZK充当着多重角色，比如注册中心、选举等；再比如说，我公司目前很多项目都是Dubbo，都是需要基于ZK实现服务发现和注册。

另外，ZK内其实也有很多优秀的算法和设计思想，熟悉ZK源码，也可以提升自己的“内功”。

如何快速入门Zookeeper呢？最简单的方式就是直接看 Zookeeper官网 啦！建议读者多参考官方文档和博客内容一起食用，效果更佳噢～

Zookeeper的 Logo 看起来就像个“铲屎官”，服务动物园内的动物们。

“A Distributed Coordination Service for Distributed Applications”，这是摘取官方的解释，我们可以得知Zookeeper 是一个为 分布式框架 提供协调服务的东东。

举些例子，有哪些分布式框架使用Zookeeper：

ZK的作用不止上面几个，其实还可以做到负载均衡、统一配置、分布式队列等，但使用场景相对少，企业级系统中，会使用其他更加专业的框架组件。

分布式锁、注册中心、Leader选举将会是ZK系列中，重点分享的内容，敬请期待哈~

在ZK中，需要先了解一些专业名词的概念，但不会一下子都列出来，当之后遇到的时候，再重点分析

在ZK集群中，会分为 Leader 、 Follower 和 Observer 角色。

Leader作为集群的大佬，承担写请求和部分读请求；Follower作为Leader的小弟，将会承担部分读请求，当接收到写请求的时候会转发给Leader，由Leader处理写请求；Observer就有点特殊，Observer节点不参与选举和消息过半机制，这个不清楚的读者可以暂时有个记忆就行，之后遇到会重点说明。

实际上，节点只分为持久节点和临时节点，但有些场景需要保证顺序，所以就会在持久或临时节点的基础上，添加序号（递增的方式），形成持久顺序节点和临时顺序节点。</br> 那么什么是持久节点，什么是临时节点呢？最直观的一个现象就是，每个ZK客户端连接ZK集群后，都会产生一个节点，如果ZK客户端下线后，节点还存在的就是持久节点，若ZK客户端下线后节点也随着消失，那么该节点就是临时节点。

在ZK客户端启动前，可以自定义监听回调函数，这个有什么作用呢？客户端启动后会将监听事件发送给Zookeeper集群，Zookeeper集群中有一个用于记录监听事件的列表，当客户端监听的目录节点发生变化，如节点数据变更、节点增删等，就会通过ZK集群的监听列表，找到对应的客户端回调监听函数，那么客户端这边就可以根据业务场景，做出相应的动作。

ZAB协议的全称是：ZooKeeper Atomic Broadcast。ZAB是Zookeeper保证数据一致性的核心算法。借鉴了Paxos算法的思想，特地为Zookeeper设计的支持崩溃恢复的原子广播协议。其包括两种基本模式： 消息广播 和 崩溃恢复

消息广播指的是，集群中只有一个Leader处理写请求，并将写请求的事件广播给所有Follower，且能够保证数据不丢失。（也就是说，消息的写入是原子性的，因为只能有leader写入）

崩溃恢复指的是，当ZK集群刚启动还没选举出Leader或Leader因故障、重启、网络等原因的时候，ZAB协议会进入崩溃恢复模式，其目的就是为了选举新的Leader，且保证新Leader的数据是最新的，这样就能够避免因为Leader故障而导致单点丢失消息的情况，至于ZAB具体的原理，各位可以先看下以下参考文章，后续有机会我再专门写一篇关于 ZAB 协议的文章~

ZAB 协议参考文章

ZK内的数据模型结构和Unix文件系统非常相似，是一个有层级关系的树形数据结构。在ZK内，树形的数据结构使用称为ZNode节点保存数据，ZNode是ZK中数据结构最小单元，不仅能够保存数据，还能挂载子节点，形成一个有层次关系的树。

值得注意的是，ZNode的创建是纯内存操作的，所以速度很快，然后在ZK内部会定期将ZNode的数据持久化到磁盘上。

众所周知，在实际的企业应用，面对高并发的场景下，肯定是不能单节点部署，而是通过集群部署保证 高并发、高性能、高可用 （简称三高）。

高性能 ：由于ZNode节点是纯内存操作，只要ZK部署在高配置的服务器中，三台ZK服务器抗住每秒几万的请求都是没问题的。 高可用 ：只要部署奇数的服务器集群（比如3台、5台、11台机器），只要不超过一半的服务器宕机，都能保证ZK集群可用。 高并发 ：因为ZNode是纯内存操作，所以在写数据的时候，速度是很快；而ZK集群中Leader和Follower节点都能处理读请求，所以ZK集群高并发能力是很强的。

基于ZAB协议，写请求统一由Leader服务器处理，然后由Leader将写数据的请求广播给其他Follower。

但会不会由于种种原因，如网络波动、Leader脑裂、Follower宕机等，导致消息不一致？

实际上，在ZK中采用2PC两阶段提交的思想，结合ZAB消息广播保证数据一致性。值得注意的是，Zookeeper只能保证最终一致性，并不能保证强一致性

那么具体是怎么保证数据最终一致性的呢？感兴趣的读者可以看下我另外一篇拙作TODO

参考资料：

《从Paxos到Zookeeper分布式一致性原理与实践》

如果觉得文章不错的话，麻烦点个赞哈，你的鼓励就是我的动力！对于文章有哪里不清楚或者有误的地方，欢迎在评论区留言~

[TOC]

我们已经知道Kafka的集群由n个的broker所组成，每个broker就是一个kafka的实例或者称之为kafka的服务。其实控制器也是一个broker，控制器也叫leader broker。

他除了具有一般broker的功能外，还负责分区leader的选取，也就是负责选举partition的leader replica。

kafka每个broker启动的时候，都会实例化一个KafkaController，并将broker的id注册到zookeeper，集群在启动过程中，通过选举机制选举出其中一个broker作为leader，也就是前面所说的控制器。

包括集群启动在内，有三种情况触发控制器选举：

1、集群启动

2、控制器所在代理发生故障

3、zookeeper心跳感知，控制器与自己的session过期

按照惯例，先看图。我们根据下图来讲解集群启动时，控制器选举过程。

假设此集群有三个broker，同时启动。

（一）3个broker从zookeeper获取/controller临时节点信息。/controller存储的是选举出来的leader信息。此举是为了确认是否已经存在leader。

（二）如果还没有选举出leader，那么此节点是不存在的，返回-1。如果返回的不是-1，而是leader的json数据，那么说明已经有leader存在，选举结束。

（三）三个broker发现返回-1，了解到目前没有leader，于是均会触发向临时节点/controller写入自己的信息。最先写入的就会成为leader。

（四）假设broker 0的速度最快，他先写入了/controller节点，那么他就成为了leader。而broker1、broker2很不幸，因为晚了一步，他们在写/controller的过程中会抛出ZkNodeExistsException，也就是zk告诉他们，此节点已经存在了。

经过以上四步，broker 0成功写入/controller节点，其它broker写入失败了，所以broker 0成功当选leader。

此外zk中还有controller_epoch节点，存储了leader的变更次数，初始值为0，以后leader每变一次，该值+1。所有向控制器发起的请求，都会携带此值。如果控制器和自己内存中比较，请求值小，说明kafka集群已经发生了新的选举，此请求过期，此请求无效。如果请求值大于控制器内存的值，说明已经有新的控制器当选了，自己已经退位，请求无效。kafka通过controller_epoch保证集群控制器的唯一性及操作的一致性。

由此可见，Kafka控制器选举就是看谁先争抢到/controller节点写入自身信息。

控制器的初始化，其实是初始化控制器所用到的组件及监听器，准备元数据。

前面提到过每个broker都会实例化并启动一个KafkaController。KafkaController和他的组件关系，以及各个组件的介绍如下图：

图中箭头为组件层级关系，组件下面还会再初始化其他组件。可见控制器内部还是有些复杂的，主要有以下组件：

1、ControllerContext，此对象存储了控制器工作需要的所有上下文信息，包括存活的代理、所有主题及分区分配方案、每个分区的AR、leader、ISR等信息。

2、一系列的listener，通过对zookeeper的监听，触发相应的操作，**的框的均为listener

3、分区和副本状态机，管理分区和副本。

4、当前代理选举器ZookeeperLeaderElector，此选举器有上位和退位的相关回调方法。

5、分区leader选举器，PartitionLeaderSelector

6、主题删除管理器，TopicDeletetionManager

7、leader向broker批量通信的ControllerBrokerRequestBatch。缓存状态机处理后产生的request，然后统一发送出去。

8、控制器平衡操作的KafkaScheduler，仅在broker作为leader时有效。

Kafka集群的一些重要信息都记录在ZK中，比如集群的所有代理节点、主题的所有分区、分区的副本信息(副本集、主副本、同步的副本集)。每个broker都有一个控制器，为了管理整个集群Kafka选利用zk选举模式，为整个集群选举一个“中央控制器”或”主控制器“，控制器其实就是一个broker节点，除了一般broker功能外，还具有分区首领选举功能。中央控制器管理所有节点的信息，并通过向ZK注册各种监听事件来管理整个集群节点、分区的leader的选举、再平衡等问题。外部事件会更新ZK的数据，ZK中的数据一旦发生变化，控制器都要做不同的响应处理。

故障转移其实就是leader所在broker发生故障，leader转移为其他的broker。转移的过程就是重新选举leader的过程。

重新选举leader后，需要为该broker注册相应权限，调用的是ZookeeperLeaderElector的onControllerFailover()方法。在这个方法中初始化和启动了一系列的组件来完成leader的各种操作。具体如下，其实和控制器初始化有很大的相似度。

1、注册分区管理的相关监听器

2、注册主题管理的相关监听

3、注册代理变化监听器

4、重新初始化ControllerContext，

5、启动控制器和其他代理之间通信的ControllerChannelManager

6、创建用于删除主题的TopicDeletionManager对象,并启动。

7、启动分区状态机和副本状态机

8、轮询每个主题，添加监听分区变化的PartitionModificationsListener

9、如果设置了分区平衡定时操作，那么创建分区平衡的定时任务，默认300秒检查并执行。

除了这些组件的启动外，onControllerFailover方法中还做了如下操作：

1、/controller_epoch值+1，并且更新到ControllerContext

2、检查是否出发分区重分配，并做相关操作

3、检查需要将优先副本选为leader，并做相关操作

4、向kafka集群所有代理发送更新元数据的请求。

下面来看leader权限被取消时，调用的方法onControllerResignation

1、该方法中注销了控制器的权限。取消在zookeeper中对于分区、副本感知的相应监听器的监听。

2、关闭启动的各个组件

3、最后把ControllerContext中记录控制器版本的数值清零，并设置当前broker为RunnignAsBroker，变为普通的broker。

通过对控制器启动过程的学习，我们应该已经对kafka工作的原理有了了解， 核心是监听zookeeper的相关节点，节点变化时触发相应的操作 。

有新的broker加入集群时，称为代理上线。反之，当broker关闭，推出集群时，称为代理下线。

代理上线：

1、新代理启动时向/brokers/ids写数据

2、BrokerChangeListener监听到变化。对新上线节点调用controllerChannelManageraddBroker()，完成新上线代理网络层初始化

3、调用KafkaControlleronBrokerStartup()处理

35恢复因新代理上线暂停的删除主题操作线程

代理下线：

1、查找下线节点集合

2、轮询下线节点，调用controllerChannelManagerremoveBroker()，关闭每个下线节点网络连接。清空下线节点消息队列，关闭下线节点request请求

3、轮询下线节点，调用KafkaControlleronBrokerFailure处理

4、向集群全部存活代理发送updateMetadataRequest请求

顾名思义，协调器负责协调工作。本节所讲的协调器，是用来协调消费者工作分配的。简单点说，就是消费者启动后，到可以正常消费前，这个阶段的初始化工作。消费者能够正常运转起来，全有赖于协调器。

主要的协调器有如下两个：

1、消费者协调器（ConsumerCoordinator）

2、组协调器（GroupCoordinator）

kafka引入协调器有其历史过程，原来consumer信息依赖于zookeeper存储，当代理或消费者发生变化时，引发消费者平衡，此时消费者之间是互不透明的，每个消费者和zookeeper单独通信，容易造成羊群效应和脑裂问题。

为了解决这些问题，kafka引入了协调器。服务端引入组协调器（GroupCoordinator），消费者端引入消费者协调器（ConsumerCoordinator）。每个broker启动的时候，都会创建GroupCoordinator实例，管理部分消费组（集群负载均衡）和组下每个消费者消费的偏移量（offset）。每个consumer实例化时，同时实例化一个ConsumerCoordinator对象，负责同一个消费组下各个消费者和服务端组协调器之前的通信。如下图：

消费者协调器，可以看作是消费者做操作的代理类（其实并不是），消费者很多操作通过消费者协调器进行处理。

消费者协调器主要负责如下工作：

1、更新消费者缓存的MetaData

2、向组协调器申请加入组

3、消费者加入组后的相应处理

4、请求离开消费组

5、向组协调器提交偏移量

6、通过心跳，保持组协调器的连接感知。

7、被组协调器选为leader的消费者的协调器，负责消费者分区分配。分配结果发送给组协调器。

8、非leader的消费者，通过消费者协调器和组协调器同步分配结果。

消费者协调器主要依赖的组件和说明见下图：

可以看到这些组件和消费者协调器担负的工作是可以对照上的。

组协调器负责处理消费者协调器发过来的各种请求。它主要提供如下功能：

组协调器在broker启动的时候实例化，每个组协调器负责一部分消费组的管理。它主要依赖的组件见下图：

这些组件也是和组协调器的功能能够对应上的。具体内容不在详述。

下图展示了消费者启动选取leader、入组的过程。

消费者入组的过程，很好的展示了消费者协调器和组协调器之间是如何配合工作的。leader consumer会承担分区分配的工作，这样kafka集群的压力会小很多。同组的consumer通过组协调器保持同步。消费者和分区的对应关系持久化在kafka内部主题。

消费者消费时，会在本地维护消费到的位置（offset），就是偏移量，这样下次消费才知道从哪里开始消费。如果整个环境没有变化，这样做就足够了。但一旦消费者平衡操作或者分区变化后，消费者不再对应原来的分区，而每个消费者的offset也没有同步到服务器，这样就无法接着前任的工作继续进行了。

因此只有把消费偏移量定期发送到服务器，由GroupCoordinator集中式管理，分区重分配后，各个消费者从GroupCoordinator读取自己对应分区的offset，在新的分区上继续前任的工作。

下图展示了不提交offset到服务端的问题：

开始时，consumer 0消费partition 0 和1，后来由于新的consumer 2入组，分区重新进行了分配。consumer 0不再消费partition2，而由consumer 2来消费partition 2，但由于consumer之间是不能通讯的，所有consumer2并不知道从哪里开始自己的消费。

因此consumer需要定期提交自己消费的offset到服务端，这样在重分区操作后，每个consumer都能在服务端查到分配给自己的partition所消费到的offset，继续消费。

由于kafka有高可用和横向扩展的特性，当有新的分区出现或者新的消费入组后，需要重新分配消费者对应的分区，所以如果偏移量提交的有问题，会重复消费或者丢消息。偏移量提交的时机和方式要格外注意！！

1、自动提交偏移量

设置 enableautocommit为true，设定好周期，默认5s。消费者每次调用轮询消息的poll() 方法时，会检查是否超过了5s没有提交偏移量，如果是，提交上一次轮询返回的偏移量。

这样做很方便，但是会带来重复消费的问题。假如最近一次偏移量提交3s后，触发了再均衡，服务器端存储的还是上次提交的偏移量，那么再均衡结束后，新的消费者会从最后一次提交的偏移量开始拉取消息，此3s内消费的消息会被重复消费。

2、手动提交偏移量

设置 enableautocommit为false。程序中手动调用commitSync()提交偏移量，此时提交的是poll方法返回的最新的偏移量。

commitSync()是同步提交偏移量，主程序会一直阻塞，偏移量提交成功后才往下运行。这样会限制程序的吞吐量。如果降低提交频次，又很容易发生重复消费。

这里我们可以使用commitAsync()异步提交偏移量。只管提交，而不会等待broker返回提交结果

commitSync只要没有发生不可恢复错误，会进行重试，直到成功。而commitAsync不会进行重试，失败就是失败了。commitAsync不重试，是因为重试提交时，可能已经有其它更大偏移量已经提交成功了，如果此时重试提交成功，那么更小的偏移量会覆盖大的偏移量。那么如果此时发生再均衡，新的消费者将会重复消费消息。