Zookeeper与Kafka

Zookeeper与Kafka

1 分布式协调技术

在给大家介绍ZooKeeper之前先来给大家介绍一种技术——分布式协调技术。那么什么是分布式协调技术？那么我来告诉大家，其实分布式协调技术主要用来解决分布式环境当中多个进程之间的同步控制，让他们有序的去访问某种临界资源，防止造成”脏数据”的后果。这时，有人可能会说这个简单，写一个调度算法就轻松解决了。说这句话的人，可能对分布式系统不是很了解，所以才会出现这种误解。如果这些进程全部是跑在一台机上的话，相对来说确实就好办了，问题就在于他是在一个分布式的环境下，这时问题又来了，那什么是分布式呢？这个一两句话我也说不清楚，但我给大家画了一张图希望能帮助大家理解这方面的内容，如果觉得不对尽可拍砖，来咱们看一下这张图，如下图所示。

给大家分析一下这张图，在这图中有三台机器，每台机器各跑一个应用程序。然后我们将这三台机器通过网络将其连接起来，构成一个系统来为用户提供服务，对用户来说这个系统的架构是透明的，他感觉不到我这个系统是一个什么样的架构。那么我们就可以把这种系统称作一个分布式系统。

那我们接下来再分析一下，在这个分布式系统中如何对进程进行调度，我假设在第一台机器上挂载了一个资源，然后这三个物理分布的进程都要竞争这个资源，但我们又不希望他们同时进行访问，这时候我们就需要一个协调器，来让他们有序的来访问这个资源。这个协调器就是我们经常提到的那个锁，比如说”进程-1”在使用该资源的时候，会先去获得锁，”进程1”获得锁以后会对该资源保持独占，这样其他进程就无法访问该资源，”进程1”用完该资源以后就将锁释放掉，让其他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。这个分布式锁也就是我们分布式协调技术实现的核心内容，那么如何实现这个分布式呢，那就是我们后面要讲的内容。

1.1 分布式锁的实现

1.1.1 面临的问题

在看了上图所示的分布式环境之后，有人可能会感觉这不是很难。无非是将原来在同一台机器上对进程调度的原语，通过网络实现在分布式环境中。是的，表面上是可以这么说。但是问题就在网络这，在分布式系统中，所有在同一台机器上的假设都不存在：因为网络是不可靠的。

比如，在同一台机器上，你对一个服务的调用如果成功，那就是成功，如果调用失败，比如抛出异常那就是调用失败。但是在分布式环境中，由于网络的不可靠，你对一个服务的调用失败了并不表示一定是失败的，可能是执行成功了，但是响应返回的时候失败了。还有，A和B都去调用C服务，在时间上 A还先调用一些，B后调用，那么最后的结果是不是一定A的请求就先于B到达呢？这些在同一台机器上的种种假设，我们都要重新思考，我们还要思考这些问题给我们的设计和编码带来了哪些影响。还有，在分布式环境中为了提升可靠性，我们往往会部署多套服务，但是如何在多套服务中达到一致性，这在同一台机器上多个进程之间的同步相对来说比较容易办到，但在分布式环境中确实一个大难题。

所以分布式协调远比在同一台机器上对多个进程的调度要难得多，而且如果为每一个分布式应用都开发一个独立的协调程序。一方面，协调程序的反复编写浪费，且难以形成通用、伸缩性好的协调器。另一方面，协调程序开销比较大，会影响系统原有的性能。所以，急需一种高可靠、高可用的通用协调机制来用以协调分布式应用。

1.1.2 分布式锁的实现者

目前，在分布式协调技术方面做得比较好的就是Google的Chubby还有Apache的ZooKeeper他们都是分布式锁的实现者。有人会问既然有了Chubby为什么还要弄一个ZooKeeper，难道Chubby做得不够好吗？不是这样的，主要是Chbby是非开源的，Google自家用。后来雅虎模仿Chubby开发出了ZooKeeper，也实现了类似的分布式锁的功能，并且将ZooKeeper作为一种开源的程序捐献给了Apache，那么这样就可以使用ZooKeeper所提供锁服务。而且在分布式领域久经考验，它的可靠性，可用性都是经过理论和实践的验证的。所以我们在构建一些分布式系统的时候，就可以以这类系统为起点来构建我们的系统，这将节省不少成本，而且bug也将更少。

2 ZooKeeper概述

ZooKeeper 是一个分布式的，开放源码的分布式应用程序协调服务，是 Google 的 Chubby 一个开源的实现。它提供了简单原始的功能，分布式应用可以基于它实现更高级的服务，比如分布式同步，配置管理，集群管理，命名管理，队列管理。它被设计为易于编程，使用文件系统目录树作为数据模型。服务端跑在 java 上，提供 java 和 C 的客户端 API 众所周知，协调服务非常容易出错，但是却很难恢复正常，例如，协调服务很容易处于竞态以至于出现死锁。我们设计 ZooKeeper 的目的是为了减轻分布式应用程序所承担的协调任务，ZooKeeper 是集群的管理者，监视着集群中各节点的状态，根据节点提交的反馈进行下一步合理的操作。最终，将简单易用的接口和功能稳定，性能高效的系统提供给用户。

前面提到了那么多的服务，比如分布式锁、配置维护、组服务等，那它们是如何实现的呢，我相信这才是大家关心的东西。ZooKeeper在实现这些服务时，首先它设计一种新的数据结构——Znode，然后在该数据结构的基础上定义了一些原语，也就是一些关于该数据结构的一些操作。有了这些数据结构和原语还不够，因为我们的ZooKeeper是工作在一个分布式的环境下，我们的服务是通过消息以网络的形式发送给我们的分布式应用程序，所以还需要一个通知机制——Watcher机制。那么总结一下，ZooKeeper所提供的服务主要是通过：数据结构+原语+watcher机制，三个部分来实现的。那么我就从这三个方面，给大家介绍一下ZooKeeper。

3 ZooKeeper数据模型

3.1 ZooKeeper数据模型Znode

ZooKeeper拥有一个层次的命名空间，这个和标准的文件系统非常相似，如下图所示。

从图中我们可以看出ZooKeeper的数据模型，在结构上和标准文件系统的非常相似，都是采用这种树形层次结构，ZooKeeper树中的每个节点被称为—Znode。和文件系统的目录树一样，ZooKeeper树中的每个节点可以拥有子节点。但也有不同之处：

(1) 引用方式

Zonde通过路径引用，如同Unix中的文件路径。路径必须是绝对的，因此他们必须由斜杠字符来开头。除此以外，他们必须是唯一的，也就是说每一个路径只有一个表示，因此这些路径不能改变。在ZooKeeper中，路径由Unicode字符串组成，并且有一些限制。字符串”/zookeeper”用以保存管理信息，比如关键配额信息。

(2) Znode结构

ZooKeeper命名空间中的Znode，兼具文件和目录两种特点。既像文件一样维护着数据、元信息、ACL、时间戳等数据结构，又像目录一样可以作为路径标识的一部分。图中的每个节点称为一个Znode。每个Znode由3部分组成:

① stat：此为状态信息, 描述该Znode的版本, 权限等信息

② data：与该Znode关联的数据

③ children：该Znode下的子节点

ZooKeeper虽然可以关联一些数据，但并没有被设计为常规的数据库或者大数据存储，相反的是，它用来管理调度数据，比如分布式应用中的配置文件信息、状态信息、汇集位置等等。这些数据的共同特性就是它们都是很小的数据，通常以KB为大小单位。ZooKeeper的服务器和客户端都被设计为严格检查并限制每个Znode的数据大小至多1M，但常规使用中应该远小于此值。

(3) 数据访问

ZooKeeper中的每个节点存储的数据要被原子性的操作。也就是说读操作将获取与节点相关的所有数据，写操作也将替换掉节点的所有数据。另外，每一个节点都拥有自己的ACL(访问控制列表)，这个列表规定了用户的权限，即限定了特定用户对目标节点可以执行的操作。

(4) 节点类型

ZooKeeper中的节点有两种，分别为临时节点和永久节点。节点的类型在创建时即被确定，并且不能改变。

① 临时节点：该节点的生命周期依赖于创建它们的会话。一旦会话(Session)结束，临时节点将被自动删除，当然可以也可以手动删除。虽然每个临时的Znode都会绑定到一个客户端会话，但他们对所有的客户端还是可见的。另外，ZooKeeper的临时节点不允许拥有子节点。

② 永久节点：该节点的生命周期不依赖于会话，并且只有在客户端显示执行删除操作的时候，他们才能被删除。

(5) 顺序节点

当创建Znode的时候，用户可以请求在ZooKeeper的路径结尾添加一个递增的计数。这个计数对于此节点的父节点来说是唯一的，它的格式为”%10d”(10位数字，没有数值的数位用0补充，例如”0000000001”)。当计数值大于232-1时，计数器将溢出。

(6) 观察

客户端可以在节点上设置watch，我们称之为监视器。当节点状态发生改变时(Znode的增、删、改)将会触发watch所对应的操作。当watch被触发时，ZooKeeper将会向客户端发送且仅发送一条通知，因为watch只能被触发一次，这样可以减少网络流量。

3.2 ZooKeeper中的时间

致使ZooKeeper节点状态改变的每一个操作都将使节点接收到一个Zxid格式的时间戳，并且这个时间戳全局有序。也就是说，每个对节点的改变都将产生一个唯一的Zxid。如果Zxid1的值小于Zxid2的值，那么Zxid1所对应的事件发生在Zxid2所对应的事件之前。实际上，ZooKeeper的每个节点维护者三个Zxid值，为别为：cZxid、mZxid、pZxid。

① cZxid：是节点的创建时间所对应的Zxid格式时间戳。

② mZxid：是节点的修改时间所对应的Zxid格式时间戳。

③ pZxid：是与该节点的子节点（或该节点）的最近一次创建 / 删除的时间戳对应

实现中Zxid是一个64为的数字，它高32位是epoch用来标识leader关系是否改变，每次一个leader被选出来，它都会有一个新的epoch。低32位是个递增计数。 (2) 版本号

对节点的每一个操作都将致使这个节点的版本号增加。每个节点维护着三个版本号，他们分别为：

① version：节点数据版本号
② cversion：子节点版本号
③ aversion：节点所拥有的ACL版本号

3.3 ZooKeeper节点属性

通过前面的介绍，我们可以了解到，一个节点自身拥有表示其状态的许多重要属性，如下图所示。

4 ZooKeeper服务中操作

在ZooKeeper中有9个基本操作，如下图所示：

更新ZooKeeper操作是有限制的。delete或setData必须明确要更新的Znode的版本号，我们可以调用exists找到。如果版本号不匹配，更新将会失败。

更新ZooKeeper操作是非阻塞式的。因此客户端如果失去了一个更新(由于另一个进程在同时更新这个Znode)，他可以在不阻塞其他进程执行的情况下，选择重新尝试或进行其他操作。

尽管ZooKeeper可以被看做是一个文件系统，但是处于便利，摒弃了一些文件系统地操作原语。因为文件非常的小并且使整体读写的，所以不需要打开、关闭或是寻地的操作。

5 监听机制

5.1 watch触发器

(1) watch概述

ZooKeeper可以为所有的读操作设置watch，这些读操作包括：exists()、getChildren()及getData()。watch事件是一次性的触发器，当watch的对象状态发生改变时，将会触发此对象上watch所对应的事件。watch事件将被异步地发送给客户端，并且ZooKeeper为watch机制提供了有序的一致性保证。理论上，客户端接收watch事件的时间要快于其看到watch对象状态变化的时间。

(2) watch类型

ZooKeeper所管理的watch可以分为两类：

① 数据watch(data watches)：getData和exists负责设置数据watch
② 孩子watch(child watches)：getChildren负责设置孩子watch

我们可以通过操作返回的数据来设置不同的watch：

① getData和exists：返回关于节点的数据信息
② getChildren：返回孩子列表

因此

① 一个成功的setData操作将触发Znode的数据watch

② 一个成功的create操作将触发Znode的数据watch以及孩子watch

③ 一个成功的delete操作将触发Znode的数据watch以及孩子watch

(3) watch注册与处触发

下图 watch设置操作及相应的触发器如图下图所示：

① exists操作上的watch，在被监视的Znode创建、删除或数据更新时被触发。
② getData操作上的watch，在被监视的Znode删除或数据更新时被触发。在被创建时不能被触发，因为只有Znode一定存在，getData操作才会成功。
③ getChildren操作上的watch，在被监视的Znode的子节点创建或删除，或是这个Znode自身被删除时被触发。可以通过查看watch事件类型来区分是Znode，还是他的子节点被删除：NodeDelete表示Znode被删除，NodeDeletedChanged表示子节点被删除。

Watch由客户端所连接的ZooKeeper服务器在本地维护，因此watch可以非常容易地设置、管理和分派。当客户端连接到一个新的服务器时，任何的会话事件都将可能触发watch。另外，当从服务器断开连接的时候，watch将不会被接收。但是，当一个客户端重新建立连接的时候，任何先前注册过的watch都会被重新注册。

(4) 需要注意的几点

Zookeeper的watch实际上要处理两类事件：

① 连接状态事件(type=None, path=null)

这类事件不需要注册，也不需要我们连续触发，我们只要处理就行了。

② 节点事件

节点的建立，删除，数据的修改。它是one time trigger，我们需要不停的注册触发，还可能发生事件丢失的情况。

上面2类事件都在Watch中处理，也就是重载的process(Event event)

节点事件的触发，通过函数exists，getData或getChildren来处理这类函数，有双重作用：

① 注册触发事件

② 函数本身的功能

函数的本身的功能又可以用异步的回调函数来实现,重载processResult()过程中处理函数本身的的功能。

5.2 监听工作原理

ZooKeeper 的 Watcher 机制主要包括客户端线程、客户端 WatcherManager、Zookeeper 服务器三部分。客户端在向 ZooKeeper 服务器注册的同时，会将 Watcher 对象存储在客户端的 WatcherManager 当中。当 ZooKeeper 服务器触发 Watcher 事件后，会向客户端发送通知，客户端线程从 WatcherManager 中取出对应的Watcher对象来执行回调逻辑

6 ZooKeeper应用举例

为了方便大家理解ZooKeeper，在此就给大家举个例子，看看ZooKeeper是如何实现的他的服务的，我以ZooKeeper提供的基本服务分布式锁为例。

6.1 分布式锁应用场景

在分布式锁服务中，有一种最典型应用场景，就是通过对集群进行Master选举，来解决分布式系统中的单点故障。什么是分布式系统中的单点故障：通常分布式系统采用主从模式，就是一个主控机连接多个处理节点。主节点负责分发任务，从节点负责处理任务，当我们的主节点发生故障时，那么整个系统就都瘫痪了，那么我们把这种故障叫作单点故障。如下图6.1和6.2所示：

图 6.1 主从模式分布式系统 　　　　　　　　　　　　　　　
Zookeeper与Kafka - 图10
图6.2 单点故障
Zookeeper与Kafka - 图11

6.2 传统解决方案

传统方式是采用一个备用节点，这个备用节点定期给当前主节点发送ping包，主节点收到ping包以后向备用节点发送回复Ack，当备用节点收到回复的时候就会认为当前主节点还活着，让他继续提供服务。如下图所示：

当主节点挂了，这时候备用节点收不到回复了，然后他就认为主节点挂了接替他成为主节点如下图7.4所示：

但是这种方式就是有一个隐患，就是网络问题，来看一网络问题会造成什么后果，如下图所示：

也就是说我们的主节点的并没有挂，只是在回复的时候网络发生故障，这样我们的备用节点同样收不到回复，就会认为主节点挂了，然后备用节点将他的Master实例启动起来，这样我们的分布式系统当中就有了两个主节点也就是—-双Master，出现Master以后我们的从节点就会将它所做的事一部分汇报给了主节点，一部分汇报给了从节点，这样服务就全乱了。为了防止出现这种情况，我们引入了ZooKeeper，它虽然不能避免网络故障，但它能够保证每时每刻只有一个Master。我么来看一下ZooKeeper是如何实现的。

6.3 ZooKeeper解决方案

1) Master启动

在引入了Zookeeper以后我们启动了两个主节点，”主节点-A”和”主节点-B”他们启动以后，都向ZooKeeper去注册一个节点。我们假设”主节点-A”锁注册地节点是”master-00001”，”主节点-B”注册的节点是”master-00002”，注册完以后进行选举，编号最小的节点将在选举中获胜获得锁成为主节点，也就是我们的”主节点-A”将会获得锁成为主节点，然后”主节点-B”将被阻塞成为一个备用节点。那么，通过这种方式就完成了对两个Master进程的调度。

(2) Master故障

如果”主节点-A”挂了，这时候他所注册的节点将被自动删除，ZooKeeper会自动感知节点的变化，然后再次发出选举，这时候”主节点-B”将在选举中获胜，替代”主节点-A”成为主节点。

(3) Master 恢复

如果主节点恢复了，他会再次向ZooKeeper注册一个节点，这时候他注册的节点将会是”master-00003”，ZooKeeper会感知节点的变化再次发动选举，这时候”主节点-B”在选举中会再次获胜继续担任”主节点”，”主节点-A”会担任备用节点。

7 Zookeeper集群介绍

Zookeeper官方提供的架构图：

上图中每一个 Server 代表一个安装 ZooKeeper 服务的服务器。组成 ZooKeeper 服务的服务器都会在内存中维护当前的服务器状态，并且每台服务器之间都互相保持着通信。

集群间通过 Zab 协议（Zookeeper Atomic Broadcast）来保持数据的一致性。

7.1 Zookeeper集群角色介绍

最典型集群模式：Master/Slave 模式（主备模式）。在这种模式中，通常 Master 服务器作为主服务器提供写服务，其他的 Slave 服务器从服务器通过异步复制的方式获取 Master 服务器最新的数据提供读服务。

但是，在 ZooKeeper 中没有选择传统的 Master/Slave 概念，而是引入了Leader、Follower 和 Observer 三种角色。

如下图所示：

ZooKeeper 集群中的所有机器通过一个 Leader 选举过程来选定一台称为 “Leader” 的机器。

Leader 既可以为客户端提供写服务又能提供读服务。除了 Leader 外，Follower 和 Observer 都只能提供读服务。

Follower 和 Observer 唯一的区别在于 Observer 机器不参与 Leader 的选举过程，也不参与写操作的“过半写成功”策略，因此 Observer 机器可以在不影响写性能的情况下提升集群的读性能。

7.2 Zookeeper集群部署

我们在之前已经说过了，Zookeeper集群特性就是整个集群中只要有超过集群数量一半才能正常工作，那么整个集群对外就是可用的，假如有2台服务器做了一个Zookeeper集群，只要有任何一台故障，那么这个Zookeeper集群就不可用了。因为剩下的一台没有超过集群一半的数量，但是当集群数量等于3台或者以上的时候，那么坏一台机器，小于3台或以上的一半，那么整个集群还能正常使用。只有当再损坏一台的时候，才会导致集群不可用。所以当你要4台组成一个集群的时候，损坏两台还剩两台，那么2台不大于集群的一半，所以3台zookeeper集群和4台zookeeper集群损坏两台的结果都是集群不可用，所以这也就是为什么我们的zookeeper的集群数量要选择奇数台是有原因的。

Hostname	IP地址	集群环境
zk-101	10.0.0.101	node-01
zk-102	10.0.0.102	node-02
zk-103	10.0.0.103	node-03

#1、三个节点分别部署JDK1.8版本，我这边采用二进制的方式
tar xf jdk-8u241-linux-x64.tar.gz
#2、配置环境变量
vim /etc/profile
export JAVA_HOME=/usr/local/jdk1.8.0_241
export PATH=$PATH:$JAVA_HOME/bin
#3、使配置文件生效：
source /etc/profile
#4、查看JDK版本，确认是否安装成功
java -version
java version "1.8.0_241"
Java(TM) SE Runtime Environment (build 1.8.0_241-b07)
Java HotSpot(TM) 64-Bit Server VM (build 25.241-b07, mixed mode)

#5、部署zookeeper集群，上传对应的zookeeper包，并解压
[root@zk-101 src]# pwd
/usr/local/src
[root@zk-101 src]# tar xf apache-zookeeper-3.6.3-bin.tar.gz
[root@zk-101 src]# mv apache-zookeeper-3.6.3-bin ../
[root@zk-101 src]# cd ../
[root@zk-101 local]# ls
apache-zookeeper-3.6.3  bin  etc  games  include  jdk1.8.0_241  lib  lib64  libexec  sbin  share  src
[root@zk-101 local]# ln -s apache-zookeeper-3.6.3-bin/ zookeeper
#6、进入zookeeper目录
[root@zk-101 local]# cd zookeeper/
[root@zk-101 zookeeper]# cd conf/
#7、基于模板复制配置文件
[root@zk-101 conf]# ls
configuration.xsl  log4j.properties  zoo_sample.cfg
[root@zk-101 conf]# cp -a zoo_sample.cfg  zoo.cfg

#8、配置文件详解：
[root@zk-101 conf]# grep -v "^$" zoo.cfg 
tickTime=2000  #服务器与服务器之间的单次心跳检测时间间隔，单位为毫秒
initLimit=10  #集群中leader服务器与follower服务器初始连接心跳次数，即多少个2000毫秒
syncLimit=5  #leader与follower之间连接完成之后，后期检测发送和应答的心跳次数，如果该follower在设置的时间内(5*2000)不能与leader进行通信，那么此follower将被视为不可用
dataDir=/tmp/zookeeper  #自定义的zookeeper保存数据的目录
clientPort=2181  #客户端连接zookeeper服务器的端口，zookeeper会监听这个端口，接受客户端的访问请求
maxClientCnxns=60  #单个客户端IP可以和zookeeper保持的连接数
autopurge.snapRetainCount=3  #3.4.0中的新增功能：启用后，zookeeper自动清除功能会将autopurge.snapRetainCount最新快照和相应的事务日志分别保留在datadir和dataLogDir中，并删除其余部分，默认值为3，最小值为3
autopurge.purgeInterval=1  #3.4.0之后的版本中，zk提供了自动清理日志和快照文件的功能，这个参数指定了清理频率，单位是小时，需要配置一个1或更大的整数，默认是0，表示不开启自动清理功能
server.1=10.0.0.101:2888:3888 #server.服务器编号=服务器IP:LF数据同步端口:LF选举端口
server.2=10.0.0.102:2888:3888
server.3=10.0.0.103:2888:3888

#9、创建数据目录
[root@zk-101 conf]# mkdir -p /data/zookeeper
#10、创建自己的集群id，要和配置文件中的server.1 对应上
[root@zk-101 conf]# echo "1" > /data/zookeeper/myid

#11、另外两个节点的配置都一样，除了myid不同
102服务器：
[root@zk-102 conf]# echo "2" > /data/zookeeper/myid
103服务器：
[root@zk-103 ~]# echo "3" > /data/zookeeper/myid

#12、各服务器启动zookeeper
[root@zk-101 zookeeper]# cd /usr/local/zookeeper/bin/
[root@zk-101 bin]# ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
#13、验证各集群状态
101服务器：
[root@zk-101 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower
102服务器：
[root@zk-102 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: leader
103服务器：
[root@zk-103 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower

7.3 zookeeper集群选举过程

7.3.1 节点角色状态

LOOKING：寻找Leader状态，处于该状态需要进入选举流程
LEADING：领导者状态，处于该状态的节点说明角色已经是Leader了
FOLLOWING：跟随者状态，表示Leader已经选举出来，当前节点角色是follower
OBSERVER：观察者状态，表名当前节点角色是observer

7.3.2 选举ID

ZXID：每个改变zookeeper状态的操作都会形成一个对应的zxid
myid：服务器的唯一标识，通过配置myid文件指定，集群中唯一

7.3.3 leader选举过程

当集群中的zookeeper节点启动以后，会根据配置文件中指定的zookeeper节点地址进行leader选择操作，过程如下：
1、每个zookeeper都会发出投票，由于是第一次选举leader，因此每个节点都会把自己当做leader角色进行选举，每个zookeeper的投票中都会包含自己的myid和zxid，此时zookeeper 1的投票为myid为1，初始zxid有一个初始值，后期会随着数据更新而自动变化，zookeeper 2的投票为myid为2，初始zxid为初始生成的值。
2、每个节点接受并检查对方的投票信息，比如投票时间、是否状态为LOOKING状态的投票。
3、对比投票，优先检查xvid，如果xvid不一样则xvid大的为leader，如果xvid相同则继续对比myid，myid大的一方为leader
成为Leader的必要条件：Leader要具有最高的zxid。当集群的规模是n时，集群中大多数的机器（至少N/2+1）得到响应并follow选出的Leader。
心跳机制：Leader与Follower；利用PING来感知对方是否存活，当Leader无法响应PING时，将重新发起Leader选举

8 Kafka

8.1 简介

8.1.1 Kafka概述

Kafka是最初由Linkedin公司开发，是一个分布式、分区的、多副本的、多订阅者，基于zookeeper协调的分布式日志系统（也可以当做MQ系统），常见可以用于web/nginx日志、访问日志，消息服务等等，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

主要应用场景是：日志收集系统和消息系统

Kafka主要设计目标如下：

以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。
高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。
支持Kafka Server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
同时支持离线数据处理和实时数据处理。
Scale out:支持在线水平扩展

8.1.2 消息系统介绍

一个消息系统负责将数据从一个应用传递到另外一个应用，应用只需关注于数据，无需关注数据在两个或多个应用间是如何传递的。分布式消息传递基于可靠的消息队列，在客户端应用和消息系统之间异步传递消息。有两种主要的消息传递模式：点对点传递模式、发布-订阅模式。大部分的消息系统选用发布-订阅模式。Kafka就是一种发布-订阅模式。

8.1.3 点对点消息传递模式

在点对点消息系统中，消息持久化到一个队列中。此时，将有一个或多个消费者消费队列中的数据。但是一条消息只能被消费一次。当一个消费者消费了队列中的某条数据之后，该条数据则从消息队列中删除。该模式即使有多个消费者同时消费数据，也能保证数据处理的顺序。这种架构描述示意图如下：

生产者发送一条消息到queue，只有一个消费者能收到。

8.1.4 发布-订阅者消息传递模式

在发布-订阅消息系统中，消息被持久化到一个topic中。与点对点消息系统不同的是，消费者可以订阅一个或多个topic，消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费，数据被消费后不会立马删除。在发布-订阅消息系统中，消息的生产者称为发布者，消费者称为订阅者。该模式的示例图如下：

发布者发送到topic的消息，只有订阅了topic的订阅者才会收到消息。

8.2 Kafka的优点

8.2.1 解耦

在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

8.2.2 冗余（副本）

有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的”插入-获取-删除”范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

8.2.3 扩展性

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。

8.2.4 灵活性&峰值处理能力

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

8.2.5 可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

8.2.6 顺序保证

在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。

8.2.7 缓冲

在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务最高效率的执行———写入队列的处理会尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。

8.2.8 异步通信

很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

8.3 Kafka中的术语解释

8.3.1 概述

在深入理解Kafka之前，先介绍一下Kafka中的术语。下图展示了Kafka的相关术语以及之间的关系：

Zookeeper与Kafka - 图23

上图中一个topic配置了3个partition。Partition1有两个offset：0和1。Partition2有4个offset。Partition3有1个offset。副本的id和副本所在的机器的id恰好相同。

如果一个topic的副本数为3，那么Kafka将在集群中为每个partition创建3个相同的副本。集群中的每个broker存储一个或多个partition。多个producer和consumer可同时生产和消费数据。

8.3.2 broker

Kafka集群包含一个或多个服务器，服务器节点称为broker。

broker存储topic的数据。如果某topic有N个partition，集群有N个broker，那么每个broker存储该topic的一个partition。

如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。

如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。

8.3.3 Topic

每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处），类似于数据库的表名。

8.3.4 Partition（分区）

topic中的数据分割为一个或多个partition。每个topic至少有一个partition。每个partition中的数据使用多个segment文件存储。partition中的数据是有序的，不同partition间的数据丢失了数据的顺序。如果topic有多个partition，消费数据时就不能保证数据的顺序。在需要严格保证消息的消费顺序的场景下，需要将partition数目设为1。

分区的优势：
1、实现存储空间的横向扩容，即将多个kafka服务器的空间结合利用
2、提升性能，多服务器读写
3、实现高可用，分区leader分布在不同的kafka服务器，比如分区0的leader为服务器A，则服务器B和服务器C为A的follower，而分区1的leader为服务器B，则服务器A和C为服务器B的follower，而分区2的leader为C，则服务器A和B为C的follower。

8.3.5 Producer

生产者即数据的发布者，该角色将消息发布到Kafka的topic中。broker接收到生产者发送的消息后，broker将该消息追加到当前用于追加数据的segment文件中。生产者发送的消息，存储到一个partition中，生产者也可以指定数据存储的partition。

8.3.6 Consumer

消费者可以从broker中读取数据。消费者可以消费多个topic中的数据。

8.3.7 Consumer Group

每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。

8.3.8 Leader

每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的partition。

8.3.9 Follower

Follower跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower，Follower与Leader保持数据同步。如果Leader失效，则从Follower中选举出一个新的Leader。当Follower与Leader挂掉、卡住或者同步太慢，leader会把这个follower从“in sync replicas”（ISR）列表中删除，重新创建一个Follower。

8.4 Kafka部署

下载地址：

http://kafka.apache.org/downloads.html

http://mirrors.hust.edu.cn/apache/

在之前的zookeeper上部署Kafka，因为Kafka要基于zookeeper才能跑起来

Hostname	IP地址	集群环境
zk-101	10.0.0.101	node-01
zk-102	10.0.0.102	node-02
zk-103	10.0.0.103	node-03

PS：kafka也是要安装jdk的哦~
#1、下载kafka包，并解压
[root@zk-101 src]# wget https://dlcdn.apache.org/kafka/2.8.0/kafka_2.12-2.8.0.tgz
[root@zk-101 src]# tar xf kafka_2.12-2.8.0.tgz
[root@zk-101 src]# mv kafka_2.12-2.8.0 /usr/local
[root@zk-101 src]# cd /usr/local
[root@zk-101 local]# ln -s kafka_2.12-2.8.0/ kafka
#2、创建数据目录，3个节点都要创建
[root@zk-101 local]# mkdir -p /data/kafka/logs
#3、修改配置文件
[root@zk-101 local]# cd /usr/local/kafka/config/
[root@zk-101 config]# vim server.properties
主要关注：server.properties 这个文件即可，我们可以发现在目录下：
有很多文件，这里可以发现有Zookeeper文件，我们可以根据Kafka内带的zk集群来启动，但是建议使用独立的zk集群
#当前机器在集群中的唯一标识，和zookeeper的myid性质一样
broker.id=1
#当前kafka对外提供服务的端口默认是9092
listeners=PLAINTEXT://10.0.0.101:9092
#这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。
advertised.listeners=PLAINTEXT://your.host.name:9092
#这个是borker进行网络处理的线程数
num.network.threads=3
#这个是borker进行I/O处理的线程数
num.io.threads=8
#发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能
socket.send.buffer.bytes=102400
#kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘
socket.receive.buffer.bytes=102400
#这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小
socket.request.max.bytes=104857600
#消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，
#如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个
log.dirs=/data/kafka/logs
#默认的分区数，一个topic默认1个分区数
num.partitions=1
#每个数据目录用来日志恢复的线程数目
num.recovery.threads.per.data.dir=1
#默认消息的最大持久化时间，168小时，7天
log.retention.hours=168
#这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件
log.segment.bytes=1073741824
#每隔300000毫秒去检查上面配置的log失效时间
log.retention.check.interval.ms=300000
#是否启用log压缩，一般不用启用，启用的话可以提高性能
log.cleaner.enable=false
#设置zookeeper的连接端口
zookeeper.connect=10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181
#设置zookeeper的连接超时时间
zookeeper.connection.timeout.ms=6000
其他的节点配置稍微不同的地方：
102服务器：
broker.id=2
listeners=PLAINTEXT://10.0.0.102:9092
103服务器：
broker.id=3
listeners=PLAINTEXT://10.0.0.103:9092
#4、启动kafka集群
PS：启动之前我们要确认一下我们的zookeeper是否启动了
101服务器：
[root@zk-101 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower
102服务器：
[root@zk-102 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: leader
103服务器：
[root@zk-103 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower
[root@zk-101 kafka]# nohup ./bin/kafka-server-start.sh ./config/server.properties &

#5、查看日志是否启动

101服务器

102服务器

103服务器

8.4.1 创建topic

[root@zk-101 kafka]# ./bin/kafka-topics.sh --create --zookeeper 10.0.0.101:2181 --replication-factor 2 --partitions 3 --topic test01
Created topic test01.
参数说明：
--create：创建topic
--zookeeper：指定zookeeper服务器
--replication-factor：要创建的副本数
--partitions 3：要创建的分区数
--topic：指定创建的topic名字

8.4.2 查看topic副本信息

[root@zk-101 kafka]# ./bin/kafka-topics.sh --describe --zookeeper 10.0.0.101:2181 --topic test01
Topic: test01    TopicId: lmmt9rpISM6J10YM0lXSrQ    PartitionCount: 2    ReplicationFactor: 3    Configs: 
    Topic: test01    Partition: 0    Leader: 2    Replicas: 2,3,1    Isr: 2,3,1
    Topic: test01    Partition: 1    Leader: 3    Replicas: 3,1,2    Isr: 3,1,2

8.4.3 查看已经创建的topic

[root@zk-101 kafka]# ./bin/kafka-topics.sh --list --zookeeper 10.0.0.101:2181
test01

8.4.4 生产者发送消息

[root@zk-101 kafka]# ./bin/kafka-console-producer.sh --broker-list 10.0.0.101:9092 --topic test01
>hello world

8.4.5 消费者消费消息

在102服务器上消费消息
[root@zk-102 kafka]# ./bin/kafka-console-consumer.sh --bootstrap-server 10.0.0.102:9092 --from-beginning --topic test01
hello world
参数详解：
--from-beginning：读取历史未消费的数据