分布式消息通知----Kafka集群部署

Kafka简介

Kafka是一款分布式的消息发布订阅系统，具有高性能，高吞吐量的特点而被广泛应用于大数据传输场景。它是由Linkedin公司开发，使用Scale语言开发，之后成为Apache基金会的一个顶级的项目。Kafka提供了类似JMS的特性，但特性和设计上完全不同，而且它也不是基于JMS规范的实现。

Kafka的产生背景

Kafka作为一个消息系统，最初设计的目的是用作Linkedin的活动流（ActivityStream）和运营数据管理通道（pipline）。活动流数据是所有的网站对用户的使用情况做分析的时候要用到的最常规的部分，活动数据包括页面的访问量（pv），被查看内容方面的信息以及搜索内容。这种数据的处理方式通常是，先把各种活动以日志的形式写入到某种文件中，然后周期性地对这些文件进行统计分析。运营数据指的是服务器的性能数据（CPU、IO使用率、请求时间、服务日志等）

Kafka的应用场景

由于Kafka具有更好的吞吐量、内置分区、冗余、容错性的优点（Kafka每秒可以处理几十万的消息），让Kafka成为一个很优秀的大规模消息处理应用的解决方案。在企业级应用上，Kafka主要应用于以下几个方面：

1. 行为跟踪：Kafka可用于跟踪用户浏览页面及搜索内容等其他动作行为，通过发布-订阅模式记录到对应的topic中，通过后端接入大数据平台处理分析，并作进一步的实时处理，监控和分析。
2. 日志收集：日志收集方面有很多优秀的产品。比如Apache Flume，很多公司使用Kafka代理日志聚合。日志聚合表示从服务器上搜集日志，然后放到一个集中的平台（文件服务管理器）进行处理。在实际应用的开发中，log会输出到本地磁盘上，排查问题的话通过Linux命令来搞定，如果应用组成了负载均衡集群，并且集群机器有几十台，甚至更多，那么想通过日志快速定位到问题，就非常麻烦了。所以一般都会做一个日志统一搜集平台，管理log日志用于快速查找定位重要应用的问题。很多公司的套路都是，将日志数据集中到Kafka上，然后分别导入es和hdfs上，用来做实时检索分析和离线数据数据统计备份等。而另一方面，Kafka本身也提供了很好的Api用来做集成日志并且做日志收集
3. 

Kafka安装部署

单机部署

1. 先部署一台zookeeper服务器，kafka的运行需要依赖于zookeeper，启动号zookeeper服务，这里zk服务的IP为192.168.50.115默认在端口2181,提供服务

[root@util bin]# sh zkServer.
zkServer.cmd  zkServer.sh   
[root@util bin]# sh zkServer.sh start
/bin/java
ZooKeeper JMX enabled by default
Using config: /home/apache-zookeeper-3.8.2-bin/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

1
2
3
4
5
6
7
8

3. 下载3.5.1的Kafka，下载地址为：
   https://downloads.apache.org/kafka/3.5.1/kafka_2.12-3.5.1.tgz

[root@zk2 home]# wget https://downloads.apache.org/kafka/3.5.1/kafka_2.12-3.5.1.tgz
--2023-07-26 15:57:34--  https://downloads.apache.org/kafka/3.5.1/kafka_2.12-3.5.1.tgz
正在解析主机 downloads.apache.org (downloads.apache.org)... 88.99.95.219, 135.181.214.104, 2a01:4f8:10a:201a::2, ...
正在连接 downloads.apache.org (downloads.apache.org)|88.99.95.219|:443... 已连接。
已发出 HTTP 请求，正在等待回应... 200 OK
长度：106956505 (102M) [application/x-gzip]
正在保存至: “kafka_2.12-3.5.1.tgz”

100%[===================================================================================================================================>] 106,956,505 24.9KB/s 用时 63m 29s

2023-07-26 17:01:05 (27.4 KB/s) - 已保存 “kafka_2.12-3.5.1.tgz” [106956505/106956505])

[root@zk2 home]# ls
admin  java  kafka_2.12-3.5.1.tgz  zookeeper
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2. 解压刚刚下载的压缩包

[root@zk2 home]# tar -zxvf kafka_2.12-3.5.1.tgz 
kafka_2.12-3.5.1/
kafka_2.12-3.5.1/LICENSE
kafka_2.12-3.5.1/NOTICE
...
[root@zk2 home]# ls
admin  java  kafka_2.12-3.5.1  kafka_2.12-3.5.1.tgz  zookeeper
1
2
3
4
5
6
7

3. 修改配置文件，更改zookeeper.connect为部署好zookeeper的服务器ip:port

[root@zk2 home]# vim kafka_2.12-3.5.1/config/server.properties
zookeeper.connect=192.168.50.115:2181 ## 将属性值改为目标zookeeper服务器ip:port
1
2

kafka默认存储消息的消息数据的目录为/tmp//kafka-logs,可以根据自己的实际需要，在service.propoties文件中进行配置

# A comma separated list of directories under which to store log files
log.dirs=/tmp/kafka-logs
1
2

配置kafka环境变量，vim /etc/profile

#this is for kafka  enviroment
export KAFKA_HOME=/home/kafka_2.12-3.5.1
export PATH=$KAFKA_HOME/bin:$PATH
1
2
3

使环境变量生效

[root@zk3 home]# source /etc/profile
1

4. 启动服务

[root@zk2 bin]# sh kafka-server-start.sh ../config/server.properties
[2023-07-26 17:55:14,596] INFO Registered kafka:type=kafka.Log4jController MBean (kafka.utils.Log4jControllerRegistration$)
[2023-07-26 17:55:18,963] INFO Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation (org.apache.zookeeper.common.X509Util)
[2023-07-26 17:55:19,921] INFO Registered signal handlers for TERM, INT, HUP (org.apache.kafka.common.utils.LoggingSignalHandler)
...
[2023-07-26 17:55:20,915] INFO [ZooKeeperClient Kafka server] Connected. (kafka.zookeeper.ZooKeeperClient)
[2023-07-26 17:55:24,261] INFO Cluster ID = nv2qazaBQk2I3xcM0WziGQ (kafka.server.KafkaServer)
...
1
2
3
4
5
6
7
8

看到[ZooKeeperClient Kafka server] Connected启动成功
5. 另外新开一个终端，创建一个topic

[root@zk2 bin]# sh kafka-topics.sh --create --topic --quickstart-events --bootstrap-server 192.168.50.112:9092
Created topic --quickstart-events.
1
2

6. 查看刚刚创建的topic

[root@zk2 bin]# sh kafka-topics.sh --list --bootstrap-server 192.168.50.112:9092  #查看所有的topic列表
--quickstart-events
__consumer_offsets
test
[root@zk2 bin]# sh kafka-topics.sh --describe --topic --quickstart-events --bootstrap-server 192.168.50.112:9092 #查看某一个topic的具体参数
Topic: --quickstart-events	TopicId: W3k6ScaISqWEy2vaQynOPA	PartitionCount: 1	ReplicationFactor: 1	Configs: 
	Topic: --quickstart-events	Partition: 0	Leader: 0	Replicas: 0	Isr: 0
1
2
3
4
5
6
7

7. 通过这个topic创建生产者，生产数据

[root@zk2 bin]# sh kafka-console-producer.sh --topic quickstart-events --bootstrap-server 192.168.50.112:9092
>this is my first message
1
2

8. 创建消费者消费这个topic上的消息

[root@zk2 bin]# sh kafka-console-consumer.sh --topic quickstart-events --bootstrap-server 192.168.50.112:9092 
this is my first message
1
2

集群部署

参考下图，将3台Kafka注册到3台zookeeper组成的集群上。

1. 先在3台机器上搭建zookeeper集群
2. 参考单机kafka单机搭建kafka的第一个节点
   修改配置文件server.properties

broker.id=1  #每个broker对应唯一一个broker.id
listeners = PLAINTEXT://192.168.50.112:9092 #配置本机的地址，即kafka监听的ip:port
zookeeper.connect=192.168.50.114:2181,192.168.50.112:2181,192.168.50.113:2181 #zookeeper集群地址
1
2
3

3. 将第一个节点配置好的kafka节点分发到另外两个节点

[root@zk2 config]# scp -r /home/kafka_2.12-3.5.1 zk1:/home/
[root@zk2 config]# scp -r /home/kafka_2.12-3.5.1 zk3:/home/
1
2

并修改对应的配置文件信息

4. 添加控制zookeeper集群的shell脚本
   在root的bin目录下新增zk.sh集群控制脚本文件

[root@zk2 bin]# vim kafka.sh
#!/bin

case $1 in
"start"){
        for i in zk1 zk2 zk3
        do
             echo ----------kafka $i 启动----------
             ssh $i "/home/kafka_2.12-3.5.1/bin/kafka-server-start.sh -daemon /home/kafka_2.12-3.5.1/config/server.properties"
        done
}
;;
"stop"){
        for i in zk1 zk2 zk3
        do
             echo ----------kafka $i 停止----------
             ssh $i "/home/kafka_2.12-3.5.1/bin/kafka-server-stop.sh"
        done
}
;;
"status"){
        for i in zk1 zk2 zk3
        do
	        echo ----------kafka $i Status----------
	        if ping -c 1 -w 1 $i > /dev/null ; then
	                num=$( ssh $i ps -ef | grep kafka | grep -vc grep ) 
	                if [ $num -eq 0 ]; then 
	                    echo "Kafka未启动"
	                else
	                    echo "Kafka已启动"
	                fi
	            else
	                echo "kafka $i 网络断开连接，无法获取状态信息"
	            fi
        done
}
;;
"topic"){
        for i in zk1 zk2 zk3
        do
             echo ----------kafka $i Topic----------
             ssh $i "/home/kafka_2.12-3.5.1/bin/kafka-topics.sh --bootstrap-server $i:9092 --list"
        done
}
;;
esac
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

5. 赋予这个脚本权限：

[root@zk2 bin]# chmod 777 zk.sh
1

6. 启动kafuka集群，如果启动出现下面错误，说明Java环境变量没找到

----------kafka zk1 启动----------
----------kafka zk2 启动----------
/home/kafka_2.12-3.5.1/bin/kafka-run-class.sh: 第 346 行:exec: java: 未找到
----------kafka zk3 启动----------
/home/kafka_2.12-3.5.1/bin/kafka-run-class.sh: 第 346 行:exec: java: 未找到
1
2
3
4
5

问题出在：

通过ssh登陆之后会找不到JAVA_HOME ，JAVA_HOME是定义在/etc/profile 里面的。
远程登录和直接登录执行的文件是不一样的：
    /etc/profile: 当用户登录时,该文件被执行.
    /etc/bashrc: 当bash shell被打开时,该文件被执行.
ssh作为远程登录的方式进入，无法触发/etc/profile的执行，找不到JAVA_HOME，
将java的配置信息配置到bashrc的文件中去，配置步骤如下所示：

(1) 命令 vim ~/.bashrc 进入到文件；

(2) 添加配置的环境变量：
#this is for java enviroment
export JAVA_HOME=/home/java/jdk8
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

#this is for zookeeper enviroment
export ZOOKEEPER_HOME=/home/zookeeper/zookeeper-bin
export PATH=$ZOOKEEPER_HOME/bin:$PATH

#this is for kafka  enviroment
export KAFKA_HOME=/home/kafka_2.12-3.5.1
export PATH=$KAFKA_HOME/bin:$PATH


(3) 命令 ：source ~/.bashrc 更新 .bashrc

对该文件进行修改保存后，在执行kafka集群启动，就不会发生如上的问题
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

如果在启动kafka集群的时候报了如下错误
ERROR Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka.common.InconsistentClusterIdException: The Cluster ID 5wuVzKv8ST-yi888x5F5kg doesn't match stored clusterId Some(b6IB_--tRdml_53SAMp_sA) in meta.properties. The broker is trying to join the wrong cluster. Configured zookeeper.connect may be wrong.
	at kafka.server.KafkaServer.startup(KafkaServer.scala:218)
	at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:44)
	at kafka.Kafka$.main(Kafka.scala:84)
	at kafka.Kafka.main(Kafka.scala)
意思是集群id跟元数据meta.properties中存储的不一致，导致启动失败。

因此去查看meta.properties文件中的元数据信息。这个文件的存储路径是通过/config/server.properties配置文件中的log.dirs属性配置的。所以通过配置文件找到meta.properties，修改里面的cluster.id即可。
1
2
3
4
5
6
7
8
9
10

重新启动kafka集群

[root@zk2 bin]# sh kafka.sh start
----------kafka zk1 启动----------
----------kafka zk2 启动----------
----------kafka zk3 启动----------
1
2
3
4

查看每个节点的启动状态

[root@zk2 bin]# sh kafka.sh status
----------kafka zk1 Status----------
Kafka已启动
----------kafka zk2 Status----------
Kafka已启动
----------kafka zk3 Status----------
Kafka已启动
1
2
3
4
5
6
7

查看每个节点下的Topic

[root@zk2 kafka-logs]# sh kafka.sh topic
----------kafka zk1 Topic----------
__consumer_offsets
test
----------kafka zk2 Topic----------
__consumer_offsets
test
----------kafka zk3 Topic----------
__consumer_offsets
test
1
2
3
4
5
6
7
8
9
10

启动后的kafka集群