kafka高可用集群搭建
说明
这篇博文主要是为了后面的 elk 做准备，我们这里搭建一个 kafka 集群，使用2个节点，还是前面的节点。主要是为了后面做数据缓冲。
这里不对 mq 的详细进行介绍，必要会对 kafka 相关配置进行描述。
- 节点说明
| 节点 | hostname |
| 192.168.179.123 | node-5 |
| 192.168.179.124 | node-4 |
| 192.168.179.125 | node-3 |
当我们进行集群搭建的时候，要注意节点数量应该为基数，要求 可用节点数量 > 总节点数量/2，我们采用最少节点数：3台
安装配置kafka
https://mirror.bit.edu.cn/apache/kafka/2.5.0/kafka_2.12-2.5.0.tgz tar -zxvf kafka_2.12-2.5.0.tgz
vi config/server.properties
# node-5 broker.id=1 listeners=PLAINTEXT://192.168.179.123:9092 offsets.topic.replication.factor=3 transaction.state.log.replication.factor=3 transaction.state.log.min.isr=3 zookeeper.connect=192.168.179.124:2181,192.168.179.123:2181,192.168.179.125:2181 # node-4 broker.id=0 listeners=PLAINTEXT://192.168.179.124:9092 offsets.topic.replication.factor=3 transaction.state.log.replication.factor=3 transaction.state.log.min.isr=3 zookeeper.connect=192.168.179.124:2181,192.168.179.123:2181,192.168.179.125:2181 # node-3 也一样 就broker.id = 2
kafka配置说明
| key | value |
| delete.topic.enable=true | 是否允许删除topic，默认false不能手动删除 |
| broker.id=0 | 当前机器在集群中的唯一标识，和zookeeper的myid性质一样 |
| listeners = PLAINTEXT://192.168.100.151:9092 | 当前kafka服务侦听的地址和端口，端口默认是9092 |
| num.network.threads=3 | 这个是borker进行网络处理的线程数 |
| num.io.threads=8 | 这个是borker进行I/O处理的线程数 |
| socket.send.buffer.bytes=102400 | 发送缓冲区buffer大小，数据不是一下子就发送的，先会存储到缓冲区到达一定的大小后在发送，能提高性能 |
| socket.receive.buffer.bytes=102400 | kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘 |
| socket.request.max.bytes=104857600 | 这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小 |
| log.dirs= | 消息日志存放的路径 |
| num.partitions=1 | 默认的分区数，一个topic默认1个分区数 |
| num.recovery.threads.per.data.dir=1 | 每个数据目录用来日志恢复的线程数目 |
| log.retention.hours=168 | 默认消息的最大持久化时间，168小时，7天 |
| log.segment.bytes=1073741824 | 这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件 |
| log.retention.check.interval.ms=300000 | 每隔300000毫秒去检查上面配置的log失效时间 |
| log.cleaner.enable=false | 是否启用log压缩，一般不用启用，启用的话可以提高性能 |
| zookeeper.connect=node1:2181,node2:2181,node3:2181 | 设置zookeeper的连接端口 |
| broker.id=0 | 当前机器在集群中的唯一标识，和zookeeper的myid性质一样 |
| zookeeper.connection.timeout.ms=6000 | 设置zookeeper的连接超时时间 |
vi config/zookeeper.properties
为了保证 kafka 的高可用，我们需要配置 zookeeper
三台主机配置保持一致即可
dataDir=/tmp/zookeeper # 客户端连接server的端口，即对外服务端口，默认为2181 clientPort=2181 # 单个客户端与单台服务器之间的连接数的限制，是ip级别的，默认是60，如果设置为0，那么表明不作任何限制。请注意这个限制的使用范围，仅仅是单台客户端机器与单台ZK服务器之间的连接数限制，不是针对指定客户端IP，也不是ZK集群的连接数限制，也不是单台ZK对所有客户端的连接数限制 maxClientCnxns=0 admin.enableServer=false # server列表 2888为选举端口，3888为心跳端口 server.0=192.168.179.124:2888:3888 # 0表示服务器的编号 对应 dataDir 下面的 myid 文件 server.1=192.168.179.123:2888:3888 server.2=192.168.179.125:2888:3888 # ZK中的一个时间单元。ZK中所有时间都是以这个时间单元为基础，进行整数倍配置的。例如，session的最小超时时间是2*tickTime tickTime=2000 # Follower在启动过程中，会从Leader同步所有最新数据，然后确定自己能够对外服务的起始状态。Leader允许F在 initLimit时间内完成这个工作。通常情况下，我们不用太在意这个参数的设置。如果ZK集群的数据量确实很大了，F在启动的时候，从Leader上同步数据的时间也会相应变长，因此在这种情况下，有必要适当调大这个参数 initLimit=10 # 在运行过程中，Leader负责与ZK集群中所有机器进行通信，例如通过一些心跳检测机制，来检测机器的存活状态。如果L发出心跳包在syncLimit之后，还没有从F那里收到响应，那么就认为这个F已经不在线了。注意：不要把这个参数设置得过大，否则可能会掩盖一些问题 syncLimit=5
在上一步 dataDir 指定的目录下，创建 myid 文件。 直接将 kafka 的 broker.id 写入对应即可
| 节点 | hostname | myid |
| 192.168.179.123 | node-5 | echo 2 > /tmp/zookeeper/myid |
| 192.168.179.124 | node-4 | echo 0 > /tmp/zookeeper/myid |
| 192.168.179.125 | node-3 | echo 1 > /tmp/zookeeper/myid |
启动集群
所有节点都需要执行
./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties ./bin/kafka-server-start.sh -daemon ./config/server.properties
进程正常启动表示成功，查看一下
生产消息+消费消息测试
./bin/kafka-console-producer.sh --bootstrap-server 192.168.179.124:9092 --topic test-messages ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.179.124:9092 --topic test-messages --from-beginning
现在 kafka 可用确认生产消费是正常的了
容错测试集群可用性
虽然说两个节点的 kafka 正常启动了，我们还需要对他的可用性进行测试，保证到时候我们服务的一个高可用。
测试：我们创建一个 topic ，我们通过查看这个 topic 里面的 leader，如果说 leader 的 kafka 服务出现问题了，会不会切换到另一个节点，做这个测试的前提是配置好副本数，没有冗余的话是没有可用性的话可能有点绝对，这里我想表明的实际是冗余是解决可用性的方案
创建 topic
./bin/kafka-topics.sh --create --bootstrap-server 192.168.179.123:9092,192.168.179.124:9092,192.168.179.125:9092 --replication-factor 3 --partitions 1 --topic test-ha-kafka # --replication-facotor 1 两个副本 # --partitions 1 创建1个分区 # --topic 主题为test
查看 topic 的 leader
./bin/kafka-topics.sh --describe --bootstrap-server 192.168.179.123:9092,192.168.179.124:9092,192.168.179.125:9092 --topic test-ha-kafka
我们手动制造错误，看集群的容错性，我们将 broker.id = 0 的 kafka 手动停止，看看 topic 的 leader 会不会自动切换
现在已经完成了我们的高可用测试，但是我们对 kafka 的管理老是通过命令行处理非常麻烦，然后给大家介绍一下 kafka 的可视化工具： kafkatool、或者可以使用 kafka-manager
可视化kafkatool使用
下载地址：https://www.kafkatool.com/，根据自己的系统下载
这样就方便多了，还可以看到我们前面测试的 topic ，搭建好这个集群先留着，要用来优化我们的 日志系统 哦。
