zookeeper+kafka
|
作者 |
刘畅 |
|
时间 |
2021-7-22 |
操作系统: Centos7.5
|
主机名 |
IP |
软件 |
|
controlnode |
172.16.1.120 |
zookeeper-3.7.0、kafka-2.8.0 |
|
slavenode1 |
172.16.1.121 |
zookeeper-3.7.0、kafka-2.8.0 |
|
slavenode2 |
172.16.1.122 |
zookeeper-3.7.0、kafka-2.8.0 |
三台服务器分别在/etc/hosts中添加如下内容，实现hosts解析
172.16.1.120 controlnode
172.16.1.121 slavenode1
172.16.1.122 slavenode2
目录
1 zookeeper架构说明
1.1 前言
1 zookeeper最早起源于雅虎研究院的一个研究小组，当时研究人员发现，在雅虎内部有很多大型系统基本都依赖一个系统来进行分布式协同，但是这些系统往往都存在分布式单点问题。所以雅虎的开发人员就开发了一个通用的无单点问题的分布式应用程序的高性能协调服务，这就是zookeeper，现由Apache基金会开源。
2 zookeeper是一个给分布式框架提供协调服务、主从架构的分布式开源框架，主要用来解决分布式集群中应用系统的一致性问题，例如怎样避免同时操作同一数据造成脏读的问题。
3 zookeeper本质上是一个分布式的小文件存储系统，提供类似于文件系统的目录树方式的数据存储，并且可以对树中的节点进行有效的管理。
4 zookeeper提供给客户端监控存储在zk内部数据的功能，从而可以达到基于数据的集群管理。例如: 统一命名服务(dubbo)、分布式配置管理(solr的配置集中管理)，分布式消息队列(sub/pub)、分布式锁、分布式协调等功能。
5 官方文档:
https://zookeeper.apache.org/
6 zookeeper集群拓扑图
1.2 ZooKeeper服务器四种状态
looking# 服务器处于寻找Leader群首的状态
leading# 服务器作为群首(leader)时的状态
following# 服务器作为follower跟随者时的状态
observing# 服务器作为观察者时的状态
1.3 zookeeper状态说明
1 leader
(1) zookeeper集群工作的核心角色；
(2) 集群内部各个服务器的调度者；
(3) 事物请求(写操作)的唯一调度和处理者，保证集群事务处理的顺序性；对于create，setData，
delete等有写操作的请求，则需要统一转发给leader处理，leader需要决定编号、执行操作，这
个过程称为一个事务。
2 follow
(1) 处理客户端非事务(读操作)请求；
(2) 转发事务请求给leader；
(3) 参与集群leader选举投票，2n-1台可以做集群投票；
3 observer
(1) 观察者角色，观察zookeeper集群的最新状态变化并将这些状态同步过来，其对于非事务请求
可以进行独立处理，对于事务请求，则会转发给leader服务器进行处理。
(2) 不会参与任何形式的投票，只提供非事务服务，通常用于在不影响集群事务处理能力的前提下
提升集群的非事务处理能力。增加了集群并发读请求。
1.4 zookeeper特点
1 zookeeper: 一个领导者(leader)，多个跟随者(follower)组成的集群。
2 leader负责进行投票的发起和决议，更新系统状态。
3 follower用于接收客户请求并向客户端返回结果，在选举leader过程中参与投票。
4 集群中只要有半数以上的节点存活，zookeeper集群就能正常服务(比如3台zookeeper，挂一台，2 > 2/3)。
5 全局数据一致: 每个server保存一份相同的数据副本，client无论连接到哪个server，数据都是一致的。
6 更新请求顺序进行。
7 数据更新原子性，一次数据更新要么成功，要么失败。
8 zookeeper是master/slave架构，zookeeper集群中的leader不是指定而来的，而是通过选举产生的。
9 zk遵循的是CP原则，即保证一致性和网络分区容错性，但不保证可用性。
1.5 zookeeper集群特性
整个集群种只要有超过集群数量一半的zookeeper工作是正常的，那么整个集群对外就是可用的，假如有2台服务器做了一个zookeeper集群，只要有任何一台故障或宕机，那么这个zookeeper集群就不可用了，因为剩下的一台没有超过集群一半的数量，但是假如有三台zookeeper组成一个集群，那么损坏一台就还剩两台，大于3台的一半，所以损坏一台还是可以正常运行的，但是再损坏一台就只剩一台集群就不可用了。那么要是4台组成一个zookeeper集群，损坏一台集群肯定是正常的，那么损坏两台就还剩两台，那么2台不大于集群数量的一半，所以3台的zookeeper集群和4台的zookeeper集群损坏两台的结果都是集群不可用，一次类推5台和6台以及7台和8台都是同理，所以这也就是为什么集群一般都是奇数的原因。
1.6 zookeeper作用
提供文件系统来存储数据。
维护和监控存储的数据状态变化，通过监控数据状态变化达到基于数据的集群管理。
主要用来解决分布式集群中应用系统的一致性问题。
1.7 zookeeper应用场景
1 主备切换
2 节点的上下线感知
3 统一命名服务
4 状态同步服务
5 集群管理
6 分布式应用配置管理
2 kafka架构说明
2.1 前言
1 kafka背景介绍
Kafka被称为下一代分布式消息系统，是非营利性组织ASF(Apache Software Foundation，简称为ASF)基金会中的一个开源项目，比如HTTP Server、Hadoop、ActiveMQ、Tomcat等开源软件都属于Apache基金会的开源软件，类似的消息系统还有RbbitMQ、ActiveMQ、ZeroMQ，最主要的优势是其具备分布式功能、并且结合zookeeper可以实现动态扩容。
kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和java编写，kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。
官方文档:
https://kafka.apache.org/quickstart
http://kafka.apache.org/documentation.html
2 kafka特点
(1) 消息持久化能力，即使对TB级以上数据也能保证常数时间的访问性能。
(2) 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输。
(3) 支持Kafka Server间的消息分区,及分布式消费,同时保证每个partition内的消息顺序传输。
(4) 同时支持离线数据处理和实时数据处理。
(5) 支持在线水平扩展
(6) 解耦
(7) 数据冗余
(8) 可恢复性
(9) 缓冲
(10) 异步通信
3 kafka集群拓扑图
2.2 异步通信原理
1 观察者模式
(1) 观察者模式又叫发布-订阅模式；
(2) 定义对象间一种一对多的依赖关系，使得每当一个对象的状态发生改变，则所有依赖于它的对象都会得到通知并自动更新；
(3) 一个目标对象的状态发生改变，所有的依赖对象(观察者对象)都将得到通知。
2 生产消费者模式
(1) 传统模式
生产者直接将消息传递给指定的消费者；
耦合性特别高，当生产者或者消费者发生改变，都需要重写业务逻辑；
(2) 生产消费者模式
通过一个容器来解决生产者和消费者的强耦合问题，生产者和消费者彼此之间不直接通信，而是通过阻塞队列来进行通讯。
(3) 数据传递流程
生产者消费者模式，N个线程进行生产，同时N个线程进行消费，两种角色通过内存缓冲区进行通讯。
生产者负责向缓冲区里添加数据单元。
消费者负责从缓冲区里面取出数据单元。
(4) 一般遵循先进先出的原则。
2.3 缓冲区
1 解耦
假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法，那么生产者对于消费者就会产生依赖。
2 支持并发
生产者直接调用消费者的某个方法过程中函数调用是同步的。
万一消费者处理数据很慢，生产者就会白白糟蹋大好时光。
3 支持忙闲不均
缓冲区还有另一个好处。如果制造数据的速度时快时慢，缓冲区的好处就体现出来了。
当数据制造快的时候，消费者来不及处理，未处理的数据可以暂时存在缓冲区中。
等生产者的制造速度慢下来，消费者再慢慢处理掉。
2.4 消息系统原理
一个消息系统负责将数据从一个应用传递到另一个应用，应用只需关注数据，无需关注数据在两个或多个应用间是如何传递的。
1 点对点消息传递
(1) 在点对点消息系统中，消息持久化到一个队列中。此时，将有一个或多个消费者消费队列中的数据，但是一条消息只能被消费一次;
(2) 当一个消费者消费了队列中的某条数据之后，该条数据则从消息队列中删除；
(3) 该模式即使有多个消费者同时消费数据，也能保证数据处理的顺序；
(4) 基于推送模型的消息系统，由消息代理记录消费状态;
(5) 消息代理将消息推送(push)到消费者后，标记这条消息为已经被消费，但是这种方式无法很好地保证消费的处理语义。
2 发布订阅消息传递
(1) 在发布-订阅消息系统中，消息被持久化到一个topic中。
(2) 消费者可以订阅一个或多个topic, 消费者可以消费该topic中所有的数据，同一条数据可以被多个消费者消费, 数据被消费后不删除。
(3) 在发布-订阅消息系统中，消息的生产者称为发布者，消费者称为订阅者。
(4) Kafka采取拉取模型(Poll)，由自己控制消费速度，以及消费的进度，消费者可以按照任意的偏移量进行消费。
2.5 流行消息队列的消息模型比较
(1) RabbitMQ实现了AQMP协议，AQMP协议定义了消息路由规则和方式。
(2) 生产端通过路由规则发送消息到不同queue，消费端根据queue名称消费消息。
(3) 此外RabbitMQ是向消费端推送消息，订阅关系和消费状态保存在服务端。
(4) 生产端发送一条消息通过路由投递到Queue，只有一个消费者能消费到。
(5) 当RabbitMQ需要支持多订阅时，发布者发送的消息通过路由同时写到多个Queue，不同订阅组消费此消息。
(6) RabbitMQ既支持内存队列也支持持久化队列，消费端为推模型，消费状态和订阅关系由服务端负责维护，消息消费完后立即删除，不保留历史消息。所以支持多订阅时，消息会多个拷贝。
rabbitmq消息存储机制参考文档: http://www.manongjc.com/detail/6-iitanulumrmkfnr.html
2 Kafka
(1) Kafka只支持消息持久化。
(2) 消费端为拉模型，消费状态和订阅关系由客户端端负责维护，消息消费完后不会立即删除，会保留历史消息。因此支持多订阅时，消息只会存储一份就可以了。
(3) 同一个订阅组会消费topic所有消息，每条消息只会被同一个订阅组的一个消费节点消费，同一个订阅组内不同消费节点会消费不同消息。
2.6 kafka各组件说明
1 Broker
kafka集群包含一个或多个服务器，服务器节点就称为broker。
2 Topic
(1) 每条发布到Kafka集群的消息都有一个类别， 这个类别被称为Topic。
(2) 类似于数据库的表名或者ES的Index。
(3) 物理上不同Topic的消息分开存储。
(4) 逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处。
创建一个TopicA的主题，3个分区分别存储在不同的服务器，注意Topic是一个逻辑上的概念。
3 Partition&Partition Replication
(1) Partition
1) topic中数据分割为一个或多个partition。
2) 每个topic至少有一个partition, 当生产者产生数据的时候，根据分配策略, 选择分区然后将消息追加到指定的分区的末尾(队列)。
3) 每条消息都会有一个自增的编号，用于标识顺序和消息的偏移量。
4) 每个partition中的数据使用多个segment文件存储。
5) partition中的数据是有序的，不同partition间的数据丢失了数据的顺序。
6) 如果topic有多个partition，消费数据时就不能保证数据的顺序。严格保证消息的消费顺序的场景下，需要将partition数目设为1。
7) broker存储topic的数据。如果某topic有N个partition, 集群有N个broker,那么每个broker存储该topic的一个partition。
8) 如果某topic有N个partition，集群有(N+M)个broker, 那么其中有N个broker存储topic的一个partition，剩下的M个broker不存储该topic的partition数据。
9) 如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中, 尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。
(2) Partition Replication
1) 数据会存放到topic的partation中，但是有可能分区会损坏。
2) 我们需要对分区的数据进行备份(备份多少取决于你对数据的重视程度)。
3) 我们将分区的分为Leader(1)和Follower(N)，Leader负责写入和读取数据，Follower只负责备份，保证了数据的一致性。
4) 备份数设置为N，表示主(leader)+备(follower)=N(和HDFS类似)。
补充:
es备份数设置N，表示主(primary shared)+备N(shared-replication)=一个shared的数量
Kafka的topic可以划分成一个或多个partition，Partition是物理上的概念。如果一个topic的副本数设为3，那么每个partition对应还会有3个相同的副本。下图我们对TopicA的分区0,1,2分别设置了3个副本，再分别存储在broker0,1,2。
4 日志分段存储
由于生产者生产的消息会不断追加到log文件末尾，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制。它将每个Partition分为多个Segment，每个Segment对应两个文件：“.index”索引文件和“.log”数据文件。
Kafka中的消息是以主题"Topic"为基本单位进行组织的，各个主题之间相互独立。在这里主题只是一个逻辑上的抽象概念，而在实际数据文件的存储中，Kafka中的消息存储在物理上是以一个或多个分区(Partition)构成，每个分区对应本地磁盘上的一个文件夹，每个文件夹内包含了日志索引文件(".index"和".timeindex")和日志数据文件(".log")两部分。
5 Leader & Follow
(1) leader
每个partition有多个副本，其中有且仅有一个作为Leader，Leader是当前负责数据的读写的partition。
(2) Follow
1) Follower跟随Leader，所有写请求都通过Leader路由，数据变更会广播给所有Follower，Follower与Leader保持数据同步。
2) 如果Leader失效，则从Follower中选举出一个新的Leader。
3) 当follower挂掉、卡住或者同步太慢，leader会把这个follower从"in sync replicas" (ISR)列表中删除，重新创建一个Follower。
每个副本都是有角色之分的，它们会选举一个副本作为leader，其余的为follower。生产者在发送数据的时候，是直接发送到leader partition，然后follower partition自行去leader进行数据同步，消费者消费数据的时候，也是从leader中消费数据。(下图在TopicA-partition-0在broker0是leader，同理其他TopicA-partition-N也有leader)。
6 Producer & Consumer & Consumer group
(1) Producer
1) 生产者即数据的发布者，该角色将消息发布到Kafka的topic中。
2) broker接收到生产者发送的消息后，broker将该消息追加到当前用于追加数据的segment文件中。
3) 生产者发送的消息，存储到一个partition中，生产者也可以指定数据存储的partition。
(2) Consumer
消费者可以从broker中读取数据。消费者可以消费多个topic中的数据。
(3) Consumer group
1) 每个Consumer属于一个特定的Consumer Group (可为每个Consumer指定group name,若不指定group name则属于默认的group)。
2) 将多个消费者集中到一起去处理某一个Topic的数据，可以更快的提高数据的消费能力。
3) 整个消费者组共享一组偏移量(防止数据被重复读取)，因为一个Topic有多个分区。
不同组的消费者可以消费同一个partition。
Kafka通过消费者组机制同时实现了发布/订阅模型和点对点模型，一个消费组由一个或多个消费者实例组成，便于扩容与容错，一个分区不会让同一个消费者组里面的多个消费者去消费，一个消费者是可以去消费多个分区的数据，当消费者数量多于partition的数量时，多余的消费者空闲，也就是说如果只有一个partition，你在同一组启动多少个consumer都没用，partition的数量决定了此topic在同一组中被可被均衡的程度。
7 offset偏移量
(1) 可以唯一的标识一条消息
(2) 偏移量决定读取数据的位置，不会有线程安全的问题，消费者通过偏移量来决定下次读取的消息。
(3) 消息被消费之后，并不被马上删除，这样多个业务就可以重复使用kafka的消息。
(4) 我们某一个业务也可以通过修改偏移量达到重新读取消息的目的，偏移量由用户控制。
(5) 消息最终还是会被删除的，默认生命周期为1周(7*24小时)。
基于时间: 1og. retention. hours=168
基于大小: 1og.retention . bytes=1073741824# 1GB
(1) 说明
ISR: in -sync Replicas加入同步队列的副本。
我们备份数据就是防止数据丢失，当主节点挂掉时，可以启用备份节点。
leader-pull-->follower
Follower每间隔一定时间去Leader拉取数据，来保证数据的同步。
(2) ISR原理
1) 当主节点挂掉，并不是去Follower选择主, 而是从ISR中选择主。
2) 判断标准
超过10秒钟没有同步数据 # replica.lag.time.max.ms=10000
主副节点差4000条数据# rerplica.lag.max.messages=4000
3) 脏节点选举
kafka采用一种降级措施来处理，选举第一个恢复的node作为leader提供服务。以它的数据为基准，这个措施被称为脏leader选举。
2.7 zookeeper在kafka集群中的作用
Kafka主要使用ZooKeeper来保存它的元数据、监控Broker和分区的存活状态，并利用ZooKeeper来进行选举。
2.8 kafka优化
1 Partition数目
(1) 一般来说，每个partition能处理的吞吐为几MB/s (仍需要基于根据本地环境测试后获取准确指标)，增加更多的partitions意味着:
1) 更高的并行度与吞吐。
2) 可以扩展更多的(同一个consumer group中的) consumers。
3) 若是集群中有较多的brokers，则可更大程度上利用闲置的brokers。
4) 但是会造成Zookeeper的更多选举。
5) 也会在Kafka中打开更多的文件。
(2) 调整准则
1) 一般来说，若是集群较小(小于6个brokers)，则配置2 x broker数的partition数。在这里主要考虑的是之后的扩展。若是集群扩展了一倍(例如12个)，则不用担心会有partition不足的现象发生。
2) 一般来说，若是集群较大(大于12个)，则配置1 x broker数的partition数。因为这里不需要再考虑集群的扩展情况，与broker数相同的partition数已经足够应付常规场景。若有必要，则再手动调整。
3) 考虑最高峰吞吐需要的并行consumer数，调整partition的数目。若是应用场景需要有20个(同一个consumer group中的) consumer并行消费，则据此设置为20个partition。
4) 考虑producer所需的吞吐，调整partition数目 (如果producer的吞吐非常高，或是在接下来两年内都比较高，则增加partition的数目。
2 Replication factor
(1) 此参数决定的是records复制的数目，建议至少设置为2，一般是3，最高设置为4。
(2) 更高的replication factor (假设数目为N)意味着:
1) 系统更稳定(允许N-1个broker宕机【es为N】)。
2) 更多的副本(如果acks=all, 则会造成较高的延时)。
3) 系统磁盘的使用率会更高(一般若是RF为3, 则相对于RF为2时，会占据更多50%的磁盘空间)。
4) 调整准则
l 以3为起始(当然至少需要有3个brdkers，同时也不建议一个Kafka集群中节点数少于3个节点)。
l 如果replication性能成为了瓶颈或是一个issue，则建议使用一个性能更好的broker，而不是降低RF的数目。
l 永远不要在生产环境中设置RF为1。
(3) ES和kafka对比
服务node分片 副本 可同时坏机器数
es 3 3 2 2
es 3 3 1 1
服务 brokepartitionreplication可同时坏机器数
kafka 3 3 32
kafka 3 3 21
3 批量写入
为了大幅度提高producer写入吞吐量，需要定期批量写文件。
(1) 每当producer写入10000条消息时，刷数据到磁盘
log.flush.interval.messages=10000
(2) 每间隔1秒钟时间，刷数据到磁盘
log.flush.interval.ms=1000
3 安装zookeeper
分别在172.16.1.120、121、122节点上进行操作
3.1 下载zookeeper
http://zookeeper.apache.org/releases.html
https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.7.0/apache-zookeeper-3.7.0-bin.tar.gz
3.2 安装
1 安装jdk环境
# tar -xzf jdk-8u45-linux-x64.tar.gz
# mv jdk1.8.0_45/ /usr/local/jdk/
# vim /etc/profile
export JAVA_HOME=/usr/local/jdk
export CLASSPATH=$JAVA_HOME/lib/tools.jar:$JAVA_HOME/jre/lib/rt.jar
export PATH=$JAVA_HOME/bin:$PATH
# source /etc/profile
# java -version
java version "1.8.0_45"
Java(TM) SE Runtime Environment (build 1.8.0_45-b14)
Java HotSpot(TM) 64-Bit Server VM (build 25.45-b02, mixed mode)
2 解压zookeeper压缩包
# tar -xzf apache-zookeeper-3.7.0-bin.tar.gz
# mv apache-zookeeper-3.7.0-bin/ /usr/local/zookeeper/
3 修改zookeeper配置文件
# mkdir -p /usr/local/zookeeper/data/
# cd /usr/local/zookeeper/
# cp -a conf/zoo_sample.cfg conf/zoo.cfg
# grep "^[a-Z]" /usr/local/zookeeper/conf/zoo.cfg
tickTime=2000
# 服务器之间或客户端与服务器之间的单次心跳检测时间间隔，单位为毫秒
initLimit=10
# 集群中leader服务器与follower服务器第一次连接最多次数
syncLimit=5
# leader与follower之间发送和应答时间，如果该follower 在设置的时间内不能与leader 进行通信，那么此 follower 将被视为不可用。
dataDir=/usr/local/zookeeper/data
# zookeeper存储内存数据快照文件的路径，并且如果没有指定其它路径的话，数据更新的事物日志也将存储到该路径下。
# 注意：事物日志会影响zookeeper服务器的整体性能，所以建议将事物日志放置到由dataLogDir参数指定的高性能磁盘路径下。
clientPort=2181
# 客户端连接Zookeeper服务器的端口，Zookeeper会监听这个端口，接受客户端的访问请求
server.1=172.16.1.120:2888:3888
# 服务器编号=服务器IP:LF数据同步端口(leader):LF选举端口
server.2=172.16.1.121:2888:3888
server.3=172.16.1.122:2888:3888
补充:
# zookeeper默认事物日志目录
dataLogDir=<zookeeper安装目录>/data
4 为zookeeper添加唯一id标识
# echo "1" > /usr/local/zookeeper/data/myid
注意:
每个节点id要不一样，172.16.1.120节点设置为1，172.16.1.121节点设置为2，172.16.1.122节点设置为3。
3.3 将zookeeper服务加入到systemd
# useradd -M -s /sbin/nologin zookeeper
# id zookeeper
uid=1000(zookeeper) gid=1000(zookeeper) groups=1000(zookeeper)
# chown -R zookeeper.zookeeper /usr/local/zookeeper/
# vim /usr/lib/systemd/system/zookeeper.service
[Unit]
Description=Zookeeper Service unit Configuration
Requires=network.target
After=network.target
[Service]
SuccessExitStatus=143
Type=forking
Environment=JAVA_HOME=/usr/local/jdk
PIDFile=/usr/local/zookeeper/data/zookeeper_server.pid
ExecStart=/usr/local/zookeeper/bin/zkServer.sh start
ExecStop=/usr/local/zookeeper/bin/zkServer.sh stop
ExecReload=/usr/local/zookeeper/bin/zkServer.sh restart
Restart=on-failure
User=zookeeper
Group=zookeeper
[Install]
WantedBy=multi-user.target
# systemctl daemon-reload
# systemctl enable zookeeper.service
# systemctl start zookeeper.service
(1) 172.16.1.120节点端口
(2) 172.16.1.121节点端口
(3) 172.16.1.122节点端口
3.4 查看zookeeper节点状态
(1) 172.16.1.120节点
[root@controlnode ~]# /usr/local/zookeeper/bin/zkServer.sh status
(2) 172.16.1.121节点
[root@slavenode1 ~]# /usr/local/zookeeper/bin/zkServer.sh status
(3) 172.16.1.122节点
[root@slavenode2 ~]# /usr/local/zookeeper/bin/zkServer.sh status
3.5 zookeeper集群验证
在172.16.1.120节点上操作
1 连接到任意节点生成数据(这里连接到172.16.1.121节点)
[root@controlnode ~]# /usr/local/zookeeper/bin/zkCli.sh -server 172.16.1.121:2181
……
[zk: 172.16.1.121:2181(CONNECTED) 0] create /test "LiuChang"
Created /test
[zk: 172.16.1.121:2181(CONNECTED) 1] ls /
[admin, brokers, cluster, config, consumers, controller, controller_epoch, feature, isr_change_notification, latest_producer_id_block, log_dir_event_notification, test, zookeeper]
[zk: 172.16.1.121:2181(CONNECTED) 2]
2 连接zookeeper节点验证数据
(1) 连接172.16.1.122节点
[root@controlnode ~]# /usr/local/zookeeper/bin/zkCli.sh -server 172.16.1.122:2181
……
[zk: 172.16.1.122:2181(CONNECTED) 0] get /test
LiuChang
[zk: 172.16.1.122:2181(CONNECTED) 1]
(2) 连接172.16.1.120节点
[root@controlnode ~]# /usr/local/zookeeper/bin/zkCli.sh -server 172.16.1.120:2181
……
[zk: 172.16.1.120:2181(CONNECTED) 0] get /test
LiuChang
[zk: 172.16.1.120:2181(CONNECTED) 1]
4 安装kafka
在172.16.1.120、121、122节点上操作
4.1 下载kafka
http://kafka.apache.org/downloads.html
https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.8.0/kafka_2.13-2.8.0.tgz
4.2 安装
1 解压kafka包
# tar -xzf kafka_2.13-2.8.0.tgz
# mv kafka_2.13-2.8.0/ /usr/local/kafka/
2 修改kafka配置文件
# vim /usr/local/kafka/config/server.properties
broker.id=1
# kafka节点的id唯一，这里172.16.1.120节点设置为1，172.16.1.121节点设置为2，
# 172.16.1.122节点设置为3。
listeners=PLAINTEXT://172.16.1.120:9092
# kafka监听的ip地址:端口， 三个节点分别设置为172.16.1.120、121、122。
log.retention.hours=168
# 保留指定小时的日志内容，24(hour)*7(day)=168(hour)
zookeeper.connect=172.16.1.120:2181,172.16.1.121:2181,172.16.1.122:2181
# 所有的zookeeper地址
补充:
# kafka默认日志目录(消息存放目录)
log.dirs=/tmp/kafka-logs
4.3 将kafka服务加入到systemd
# useradd -M -s /sbin/nologin kafka
# id kafka
uid=1001(kafka) gid=1001(kafka) groups=1001(kafka)
# chown -R kafka.kafka /usr/local/kafka/
# vim /usr/lib/systemd/system/kafka.service
[Unit]
Description=Apache Kafka server (broker)
Requires=network.target
After=network.target zookeeper.service
[Service]
SuccessExitStatus=143
Type=simple
Environment=JAVA_HOME=/usr/local/jdk
ExecStart=/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties
ExecStop=/usr/local/kafka/bin/kafka-server-stop.sh
ExecReload=/bin/kill -HUP $MAINPID
Restart=on-failure
User=kafka
Group=kafka
[Install]
WantedBy=multi-user.target
# systemctl daemon-reload
# systemctl start kafka.service
# systemctl enable kafka.service
补充: shell命令后台守护进程启动
# /usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
(1) 172.16.1.120节点端口
(2) 172.16.1.121节点端口
(3) 172.16.1.122节点端口
4.4 验证zookeeper和kafka进程
(1) 172.16.1.120节点
(2) 172.16.1.121节点
(3) 172.16.1.122节点
4.5 kafka集群验证topic
在172.16.1.120节点上操作
1 创建topic
[root@controlnode ~]# /usr/local/kafka/bin/kafka-topics.sh --create --bootstrap-server 172.16.1.120:9092,172.16.1.121:9092,172.16.1.122:9092 --partitions 3 --replication-factor 3 --topic messagetest
参数说明:
--create --zookeeper 172.16.1.120:2181,172.16.1.121:2181,172.16.1.122:2181
# 表示连接kafka集群
--partitions 3 --replication-factor 3
# 表示创建topic的partitions数目为3，每个partition的replication为3，
# 即每个partition的partition_leader+follower=3。
2 查看topic的描述信息
[root@controlnode ~]# /usr/local/kafka/bin/kafka-topics.sh --describe --bootstrap-server 172.16.1.120:9092,172.16.1.121:9092,172.16.1.122:9092 --topic messagetest
状态说明:
messagetest有三个分区，分别为0、1、2，分区0的leader是1(broker.id)，分区0有三个副本，并且状态都为lsr(ln-sync，表示可以参加选举成为leader)。
3 获取所有topic
[root@controlnode ~]# /usr/local/kafka/bin/kafka-topics.sh --list --bootstrap-server 172.16.1.120:9092,172.16.1.121:9092,172.16.1.122:9092
4 在topic中生产消息
[root@controlnode ~]# /usr/local/kafka/bin/kafka-console-producer.sh --broker-list 172.16.1.120:9092,172.16.1.121:9092,172.16.1.122:9092 --topic messagetest
5 从topic开始位置消费消息
[root@controlnode ~]# /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server 172.16.1.120:9092,172.16.1.121:9092,172.16.1.122:9092 --topic messagetest --from-beginning
6 删除topic
[root@controlnode ~]# /usr/local/kafka/bin/kafka-topics.sh --delete --bootstrap-server 172.16.1.120:9092,172.16.1.121:9092,172.16.1.122:9092 --topic messagetest
验证topic是否被删除:
[root@controlnode ~]# /usr/local/kafka/bin/kafka-topics.sh --describe --bootstrap-server 172.16.1.120:9092,172.16.1.121:9092,172.16.1.122:9092 --topic messagetest
7 说明
(1) kafka新旧版本命令差别
--zookeeper 172.16.1.120:2181,172.16.1.121:2181,172.16.1.122:2181
# kafka旧版本命令，可能不兼容新版本
--bootstrap-server 172.16.1.120:9092,172.16.1.121:9092,172.16.1.122:9092
# kafka新版的命令
(2) 错误命令报错
# /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server 172.16.1.120:2181,172.16.1.121:2181,172.16.1.122:2181 --topic messagetest --from-beginning
错误信息:
WARN [Consumer clientId=consumer-console-consumer-91588-1, groupId=console-consumer-91588] Bootstrap broker 172.16.1.121:2181 (id: -2 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)