zookeeper和kafka简单介绍

zookeeper概念介绍 

在介绍zookeeper之前，先来介绍一下分布式协调技术，所谓分布式协调技术主要是用来解决分布式环境中多个进程之间的同步控制，让他们有序的去访问某一种共享资源，防止造成资源竞争（脑裂）的后果
 
 分布式系统，分布式系统就是在不同地域分布的多个服务器，共同组成的一个应用系统来为用户提供服务，在分布式系统中最重要的是进程的调度，这里假设有一个应用系统，在第一台机器上挂载了一个资源，然后这三个地域分布式的应用进程都要竞争这个资源，但我们又不希望多个进程同时进行访问，这个时候就需要一个协调器，来让它们有序的来访问这个资源，这个协调器就是分布式系统中经常提到的那个“锁”

 例如“进程1”在使用该资源，会先去获得这把锁，"进程1"获得锁后回对该资源保持独占，此时其他进程就无法访问该资源，"进程1"在用完该资源以后会将该锁释放掉，以便其它进程来获得锁。通过这个"锁"机制，就可以保证分布式系统中多个进程能够有序的访问该共享资源，这里把这个分布式环境下这个"锁"叫作分布式锁，分布式锁就是分布式协调技术实现的核心内容。
 
 zookeeper是一种为分布式应用所设计的高可用，高性能的开源协调服务，它提供溜了一项基本服务：分布式锁服务，同时也提供了数据的维护和管理机制，如：统一的命名服务，状态同步服务，集群管理，分布式消息列队，风不顺应用配置项的管理等等。

zookeeper应用举例 

这里以zookeeper提供的基本服务分布式锁为例子介绍。在分布式锁服务器，有一种最典型的应用场景，就是通过对集群进行master角色的选举，来解决分布式系统中的单点故障问题。所谓单点故障，就是一个主从的分布式系统中，主节点负责任务调度分发，从节点负责任务处理，当主节点发生故障时，整个应用系统也就瘫痪了，这种故障就是单点故障。
 
 解决单点故障，传统方式是采用备用节点，这个备用节点会定期向主节点发送ping包，主机点收到ping包后向备用节点发送回复Ack信息，当备用节点收到回复就会认为当前主节点运行正常，让它继续提供服务。当主节点故障，备用节点无法收到回复信息，此时备用节点就会任务主节点宕机，然后接替称为新的主节点继续提供服务。

传统解决单点故障的方法，虽然在一定程度上解决了问题，但是又一个隐患，就是网络问题，可能会存在这样一种情况：主节点并没有故障，只是在回复Ack响应时候网络发生了故障，这样备用节点就无法收到回复，那么它就会认为主节点发生了故障，备用节点将节点主节点的服务，并成为新的主节点，此时，分布式系统中就出现了两个主节点的情况，会导致分布式系统服务发生混乱。这样的话，整个分布式系统将变得不可以，为了防止这样的情况发生，就需要引入zookeeper来解决这种问题。

 ### zookeeper工作原理
 下面通过三种情形，介绍下zookeeper是如何进行工作。
 (1).Master启动
    在分布式系统中引入zookeeper以后，就看可以配置多个主节点，这里以配置两个主节点为例，假定它们是“主节点A”和“主节点B”,两个主节点都启动后，它们都会向zookeeper中注册节点信息。假设“主节点A”锁注册的节点信息是“master00001”，“主节点B”注册的节点信息是“master00002”，注册完成后会进行选举，选举又多种算法，这里以编号最小的作为选举算法，那么编号最小的节点将在选举中获胜并获得锁成为主节点，“主节点A”将会获得锁成为主节点，然后“主节点B”将被阻塞成为一个备用节点。通过这种方式zookeeper就完了了对两个master进度的调整。完成了主,备节点的分配和协作。

（2）.Master故障
    如果 “主节点A”发生故障，这个时候它在zookeeper中所注册的节点信息会被自动删除，而zookeeper会自动感知节点的变化，发现“主节点A”故障后，会再次发出选举，这时候“主节点B”将会在选举中获胜，替代“主节点A”成为新的主节点，这样就完了主,被节点的重新选举

（3）.Master恢复
   如果主节点恢复了，它会再次向zookeeper注册自身的节点信息，只不过它注册的节点信息将会变成“master00003”，而不是原来的信息。zookeeper会感知节点的变化再次发动选举，这时候“主节点B”在选举中会再次获胜继续担任“主节点”，“主节点A”会担任备用节点。

zookeeper就是通过这样的协调，调度机制如此反复的对集群进行管理和状态同步

 zookeeper集群架构

zookeeper一般是通过集群架构来提供服务，下图是zookeeper的基本架构图

 zookeeper集群主要角色又server和client，其中server又分为leader，follower和observer三个角色，每个角色信息如下：

* leader：领导者角色，主要负责投票的发起和决议，以更新系统状态
* follower：跟随者角色，用于接收客户端请求并返回结果给客户端，在选择过程中参与投票。
* observer：观察者角色，用户接收客户端的请求，并将请求转发个leader，同时同步leader状态，但不参与投票，observer目的是扩展系统，提高伸缩性。
* client： 客户端角色，用于向zookeeper发起请求。
 
  zookeeper集群中每个server在内存中存储一份数据，在zookeeper启动时，将从实例中选举一个server作为leader，leader负责处理数据更新等操作，当且仅大多数server在内存中修改数据成功时，才认为修改数据成功。

  zookeeper写的流程：客户端client首先和一个server或observer通信，发起写请求，然后server将写请求转发给leader，leader在将写请求转发给其他server，其它server在接收到写请求后写入数据并响应leader，leader在接收到大多数写成功回应后，认为数据写成功，最后响应client，完成一次写操作

安装配置zookeeper集群

环境	IP	安装应用
centos8	192.168.248.133	zookeeper Kafka
centos8	192.168.248.134	zookeeper Kafka
centos8	192.168.248.156	zookeeper Kafka

对于集群模式下的zookeeper部署，官方建议至少要三台服务器，关于服务器推荐时奇数个，以实现zookeeper集群的高可用，这里使用三台服务器进行部署下面，配置3台机器都要部署

下载JDK官方网站

安装JDK

[root@zookkafka ~]# tar zxvf jdk-8u281-linux-x64.tar.gz  -C /usr/local/

# 配置环境变量
[root@zookkafka ~]# vim /etc/profile
.................
export JAVA_HOME=/usr/local/jdk1.8.0_281
export PATH=$PATH:$JAVA_HOME/bin
export CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib.dt.jar:$CLASSPATH

[root@zookkafka ~]# source  /etc/profile
[root@zookkafka ~]# java -version
java version "1.8.0_281"
Java(TM) SE Runtime Environment (build 1.8.0_281-b09)
Java HotSpot(TM) 64-Bit Server VM (build 25.281-b09, mixed mode)

 1）.下载与安装zookeeper

zookeeper使用Java编写的，需要安装Java运行环境，从https://zookeeper.apache.org/获取zookeeper安装包

这里安装apache-zookeeper-3.6.2-bin.tar.gz

[root@zookkafka ~]# tar -zvxf apache-zookeeper-3.6.2-bin.tar.gz  -C /usr/local/
[root@zookkafka ~]# mv /usr/local/apache-zookeeper-3.6.2-bin/  /usr/local/zookeeper

 2).配置zookeeper

zookeeper的配置模板文件为/usr/local/zookeeper/conf/zoo_sample.cfg,拷贝该文件并重命名zoo.cfg，重点配置内容如下：

[root@zookkafka ~]# cd /usr/local/zookeeper/conf/
[root@zookkafka conf]# cp zoo_sample.cfg  zoo.cfg

[root@zookkafka conf]# vim zoo.cfg
tickTime=2000     #zookeeper使用时间的度量单位毫秒
initLimit=10    #10个ticktime zookeeper中follower初始连接到leader最长时间  
syncLimit=5    #5个ticktime  用来表示leader和follower之前请求和应答最大请求时间
dataDir=/testdata/zookeeper  #存储快照文件，也就是zookeeper存储的数据文件，目录时必须已经存在的 
clientPort=2181
server.1=192.168.248.133:2888:3888 #.1表示集群中第1台服务器，2888表示这个服务器与集群中leaders服务器通通讯的端口，3888执行选举时服务器之间通讯的端口
server.2=192.168.248.134:2888:3888
server.3=192.168.248.134:2888:3888

 除了修改zoo.cfg配置文件外，集群模式下还要配置一个文件myid，这个文件需要放在dataDir配置项指定 的目录下，这个文件里面只有一个数字，如果写入1表示第一个服务器，与zoo.cfg的server.1对应，依次类推

3）.启动服务

[root@zookkafka conf]# cd ..
[root@zookkafka zookeeper]# ./bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

[root@zookkafka zookeeper]# jps
10768 Jps
10702 QuorumPeerMain #这个标识代表启动成功

 Kafka基本概念  

Kafka是一种高吞吐量的分布式发布/订阅消息系统，它的最大特性就是可以实时的处理大量数据以满足各种需求场景： 比如基于Hadoop平台的数据分析，低延时的实时系统，storm/spark流式处理引擎等。
   

Kafka角色术语

*  Broker： Kafka集群包含一个或者多个服务器，每个服务器都被称为broker
*  Topic: 每条消息发布到Kafka集群的消息都有一个分类，这个类别被称为Topic（主题）
*  Producer: 指消息的产生者，负责发布消息到Kafka broker
*  Consumer： 指消息的消费者，从kafka broker拉取数据，并消费这些已发布的消息
*  Partition：partition是一个物理上的概念，每一个Topic包含一个或者多个partition，每一个partition都是一个有序的队列，其中partition中每条消息都会被分配一个有序的offset（id）
*  Consuner Group：消费者组，可以给每一个consumer指定消费者组，若不指定消费者组，则属于默认group
*  Message： 消息，通信的基本单位，每一个producer可以向一个topic发布一些消息

Kafka拓扑架构

  一个典型的Kafka集群包含若干个Producer，若干broker,若干ConsumerGroup，以及一个zookeeper集群，Kafka通过zookeeper管理集群配置，选举leader，以及在Consunmer Group发生变化时进行。Producer使用push模式将消息发布到broker,Consunmer使用pull模式从broker拉取消息

Topic与Partition

 kafka中topic是以partition的形式存放的，每一个topic都可以设置它的partition数量，partition的数量决定了组成topic的log数量。推荐partition的数量一定要大于同时运行的consumer的数量。建以partition的数量大于broker的数量，这样消息的数据就可以均匀的分布在各个broker中。
 
 topic中设置多个partition，是因为Kafka是基于文件存储的，通过配置多个partition可以将消息的内容分散存储到多个broker上，这样可以避免文件大小到达单机磁盘的上限。同时，将一个topic切分成任意多个partitions，可以保证消息的存储、消费的效率，越多的partitions可以容纳更多的consumer，可以有效的提高Kafka的吞吐率。因此将topic切分成多个partitions的好处是可以将大量的消息分成多批数据同时写到不同节点上，将写请求分担负载到各个集群节点。

Kafka消息发送机制

  每当用户往某一个topic发送数据时，数据会被hash到不同的partition，这些partition位于不同的集群节点上,使用每个消息都会被记录一个offset消息号，消费者通过这个offset号去查询读取这个消息。
    
发送消息流程=>： 首先获取topic的所以partition，若干客户端不指定partition，也没有指定key的话，使用自增长的数字取余的方式实现指定partition。这样Kafka将平均的向partition中产生数据。如果想要控制发送的partition可以指定partition或者根据key自己写算法实现。
每一条消息被发送到broker时，会根据partition规则选择被存储到哪一个partition。如果partition规则设置的合理，所以消息可以均匀的分布到不同的partition里，这样就实现了水平扩展，同时，每一条消息被append到partition中时，时顺序写入磁盘的，因此效率非常高，顺序写入磁盘效率比随机写入内存还要高，这是Kafka高吞吐率的一个重要的保证

Kafka消费机制

     kakfa中的producer和consumer采用的push、pull
的模式，producer只向broker push消息，consumer只从brokerpull消息，push和pull对于消息的生产和消费是异步进行。pull模式的一个好处是consumer可自主控制消费者消息的速率，同时consumer还可以自己控制消费消费消息的方式是批量的从broker拉取数据还是逐条消费数据。

     当生产者将数据发布到topic时，消费者通过pull的方式，定期从服务器拉取数据，从服务器拉取数据时，服务器会告诉consumer可消费消息的offset

**消费规则**
* 不同consumer gruop下的消费者可以消费partition中相同消息，相同consumer group下的消费者只能消费partition中的不同数据
* topic的partition的个数和同一个消费组的消费组个数最好一致，如果消费组个数多于partition个数，则会存在又消费组消费不到的数据。
* 服务器会记录每个consumer的在每个partition下的消费的offset，然后每次去拉取数据时，都会从上一次记录的位置开始拉取数据

Kafka消息存储机制

在存储结构上，每个partition在物理上对应一个文件夹，该文件夹下存储这个partition的所有消息和索引文件，每一个partition（目录）相当于一个大型文件被平均分配到多个大小相等的segment（段）数据文件中。

partition命名规则为topic名称+序列号，第一个partition序号从0开始，最大值为partition数量减1

每一个partition（文件夹）中又多个大小相等的segment（段）数据文件，每个segment的大小时相同的，但是每条消息的大小可以不相同，因此segment数据文件中的消息数量不一定相等。

下载安装Kafka

从Kafka官网http://kafka.apache.org/downloads获取Kafka安装包

这里安装kafka_2.10-0.10.2.2.tgz

1）.安装

[root@zookkafka ~]# tar -zxvf kafka_2.10-0.10.2.2.tgz   -C /usr/local/
[root@zookkafka ~]# mv /usr/local/kafka_2.10-0.10.2.2/ /usr/local/kafka

2）.配置Kafka集群
kafka主配置文件为/usr/local/kafka/config/server.properties,这里介绍了一些常用的配置项

[root@zookkafka ~]# cd /usr/local/kafka/

[root@localhost kafka]# vim config/server.properties
broker.id=1  #每个节点在集群中唯一标识，1标识在Kafka集群中第一台服务器只能是数字整数
listeners=PLAINTEXT://192.168.248.133:9092   #设置Kafka监听地址和端口
log.dirs=/usr/local/kafka/log #出来放日志之外还可以保存Kafka传输的数据，可以写多个后面接上“，”接路径
num.partitions=6 #用来设置新创建的topic有多少个partition（根据consumer来定）
log.retention.hours=60  #消息的保存时间还支持log.retention.minutes和log.retention.ms如果使用多个自动选择最小，推荐ms
log.segment.bytes=1073741824   #每个partition当中每个segment(段)数据文件大小,默认1G
zookeeper.connect=192.168.248.133:2181,192.168.248.134:2181,192.168.248.156:2181   #zookeeper集群所在的IP信息
auto.create.topic.enable=true  #允许自动创建topic
delete.topic.enable=true  #是否打开自动删除的功能 “不打开此功能删除只是逻辑上删除，只是删除对应配置文件和属性但是数据没有删除”

3）.启动Kafka

[root@zookkafka kafka]# nohup  ./bin/kafka-server-start.sh  config/server.properties &

#查看启动信息
[root@zookkafka kafka]# tail -f nohup.out

[root@zookkafka kafka]# jps
11417 Jps
10702 QuorumPeerMain
11134 Kafka  #正常启动

Kafka集群基本命令操作

#创建主题
bin/kafka-topics.sh --create  --zookeeper 192.168.248.133:2181,192.168.248.134:2181,192.168.248.156:2181 --replication-factor 1 --partitions 3 --topic mytopic
Created topic "mytopic".

#查看主题属性
[root@zookkafka kafka]# bin/kafka-topics.sh  --zookeeper 192.168.248.133:2181,192.168.248.134:2181,192.168.248.156:2181 --list
mytopic

[root@zookkafka kafka]# bin/kafka-topics.sh --describe  --zookeeper 192.168.248.133:2181,192.168.248.134:2181,192.168.248.156:2181  --topic mytopic
Topic:mytopic	PartitionCount:3	ReplicationFactor:1	Configs:
	Topic: mytopic	Partition: 0	Leader: 1	Replicas: 1	Isr: 1
	Topic: mytopic	Partition: 1	Leader: 2	Replicas: 2	Isr: 2
	Topic: mytopic	Partition: 2	Leader: 3	Replicas: 3	Isr: 3

生产消息

[root@zookkafka kafka]# bin/kafka-console-producer.sh --broker-list 192.168.248.133:9092,192.168.248.134:9092,192.168.248.156:9092 --topic mytopic
etst adhasihd   #交互输入消息
test
hhhh

#在另一节点从最开始查看消息--from-beginning
[root@zookkafka1 kafka]# bin/kafka-console-consumer.sh --bootstrap-server 192.168.248.133:9092,192.168.248.134:9092,192.168.248.156:9092  --topic mytopic --from-beginning
etst adhasihd
weqeq
dasdasdasfewgfdgfd
test
meilaob
qweqeq
hhhh
dasda

删除主题

[root@zookkafka kafka]# bin/kafka-topics.sh  --zookeeper 192.168.248.133:2181,192.168.248.134:2181,192.168.248.156:2181  --delete --topic mytopic
Topic mytopic is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.