【kafka】kafka单节点/集群搭建_kafka 单节点

概述

本章节将分享不同版本的kafka单节点模式和集群模式搭建。
在kafka2.8版本之前，需要依赖zookeeper服务，而在kafka2.8版本（包括）之后，可以不在依赖zookeeper服务。本章节将分kafka2.8版本之前的版本和之后的版本分别搭建单节点模式和集群模式。

服务器环境规划

实际的生产使用中，我们一般推荐搭建奇数多节点的kafka集群，如3/5/7。在本次测试中，我分别使用了1台和3台Centos7 三台服务器搭建，复用了我搭建之前k8s集群的环境，如下表。

java环境搭建

参考https://blog.csdn.net/margu_168/article/details/132598962

2.8版本

安装包下载

直接在服务器用wget下载或者用迅雷下载好了上传也行。下载地址https://kafka.apache.org/downloads

[root@k8s-m1 ~]# wget https://archive.apache.org/dist/kafka/2.8.0/kafka_2.13-2.8.0.tgz
1

单节点模式

#解压
[root@k8s-m1 ~]# tar -xvf kafka_2.13-2.8.0.tgz 
#进入解压后的目录
[root@k8s-m1 ~]# cd kafka_2.13-2.8.0/
#生成uuid
[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-storage.sh random-uuid
MJufIDcZRMmG0-brb3nRhg
# 将uuid写入配置文件中，注意要使用上一步骤中生产的uuid
[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-storage.sh format -t MJufIDcZRMmG0-brb3nRhg -c ./config/kraft/server.properties
Formatting /root/kraft-combined-logs
#启动命令
[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-server-start.sh ./config/kraft/server.properties
#后台启动
./bin/kafka-server-start.sh ./config/kraft/server.properties &
或者
./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

测试使用

#创建topic
[root@k8s-m1 kafka_2.13-2.8.0]#./bin/kafka-topics.sh --create --topic testkafka --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092 
Created topic testkafka.
#创建生产者
[root@k8s-m1 kafka_2.13-2.8.0]#./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testkafka 
#创建消费者
[root@k8s-m1 kafka_2.13-2.8.0]#./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testkafka --from-beginning
#查看topic列表
[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-topics.sh --list --bootstrap-server localhost:9092 
testkafka
#查看topic状态
[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-topics.sh --describe  --bootstrap-server localhost:9092 
Topic: testkafka        TopicId: vtjXyJpVRIWPMkSgWm6uOA PartitionCount: 1       ReplicationFactor: 1    Configs: segment.bytes=1073741824
        Topic: testkafka        Partition: 0    Leader: 1       Replicas: 1     Isr: 1
1
2
3
4
5
6
7
8
9
10
11
12
13
14

集群模式

规划的三个节点上都需要相应的安装包。
解压

[root@k8s-m1 ~]# tar -xvf kafka_2.13-2.8.0.tgz 
1

修改配置，并生产uuid进行格式

进入解压后的conf/kraft目录，修改server.properties中的nodeid，注意每个节点上的nodeid不一样，我们分别规划为1/2/3。修改controller.quorum.voters为以下格式，注意我们规划的投票端口为9093。其他Broker进行通信，传递Topic的消息端口都设置为9092。如果服务器不够，将3个broker部署在一台服务器上，需要注意端口不能冲突。

[root@k8s-m1 ~]# cd kafka_2.13-2.8.0/conf/kraft
#修改k8s-m1
[root@k8s-m1 kafka_2.13-2.8.0]# vim config/kraft/server.properties 
node.id=1
controller.quorum.voters=1@192.168.2.140:9093,2@192.168.2.141:9093,3@192.168.2.142:9093
listeners=PLAINTEXT://192.168.2.140:9092,CONTROLLER://192.168.2.140:9093
inter.broker.listener.name=PLAINTEXT
advertised.listeners=PLAINTEXT://192.168.2.140:9092
1
2
3
4
5
6
7
8

其次生成uuid，并使用生成的uuid格式化存储目录时使用的uuid（集群id）只需要一个。依次使用相同命令格式化另外两个节点。

#生成uuid
[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-storage.sh random-uuid
HXJpfi94Q8avP4wkBVRdfw
# 将uuid写入配置文件中，注意要使用上一步骤中生产的uuid
[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-storage.sh format -t HXJpfi94Q8avP4wkBVRdfw -c ./config/kraft/server.properties
Formatting /tmp/kraft-combined-logs

#k8s-m2
[root@k8s-m2 kafka_2.13-2.8.0]# ./bin/kafka-storage.sh format -t HXJpfi94Q8avP4wkBVRdfw -c ./config/kraft/server.properties
Formatting /tmp/kraft-combined-logs

#k8s-m3
[root@k8s-m2 kafka_2.13-2.8.0]# ./bin/kafka-storage.sh format -t HXJpfi94Q8avP4wkBVRdfw -c ./config/kraft/server.properties
Formatting /tmp/kraft-combined-logs
1
2
3
4
5
6
7
8
9
10
11
12
13
14

启动并进行检查
三台服务都使用以下命令在守护程序模式下启动kafka服务。同样的命令启动另外两条服务器上的kafka。

[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties
1

使用jps检查kafka是否启动

#k8s-m1
[root@k8s-m1 kafka_2.13-2.8.0]# jps
28206 Kafka
28415 Jps
#k8s-m2
[root@k8s-m2 kafka_2.13-2.8.0]# jps
20794 Kafka
3548 Jps
#k8s-m3
[root@k8s-m3 kafka_2.13-2.8.0]# jps
2034 Jps
21935 Kafka
1
2
3
4
5
6
7
8
9
10
11
12

测试使用

#创建topic
[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-topics.sh --create --topic testkafka1 --partitions 3 --replication-factor 3 --bootstrap-server 192.168.2.140:9092,192.168.2.141:9092,192.168.2.142:9092
Created topic testkafka1.
#查看topic
[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-topics.sh --describe  --topic testkafka1  --bootstrap-server 192.168.2.140:9092Topic: testkafka1       TopicId: s_AFGUSfRHWb8FSQjdwaCw PartitionCount: 3       ReplicationFactor: 3    Configs: segment.bytes=1073741824
        Topic: testkafka1       Partition: 0    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
        Topic: testkafka1       Partition: 1    Leader: 3       Replicas: 3,2,1 Isr: 3,2,1
        Topic: testkafka1       Partition: 2    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3
[root@k8s-m1 kafka_2.13-2.8.0]# 
1
2
3
4
5
6
7
8
9

模拟producer

[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-console-producer.sh --bootstrap-server 192.168.2.140:9092,192.168.2.141:9092,192.168.2.142:9092 --topic testkafka1
>hello
>kafka
>123
>abc
>
1
2
3
4
5
6

模拟consumer

[root@k8s-m1 kafka_2.13-2.8.0]# ./bin/kafka-console-consumer.sh  --bootstrap-server 192.168.2.140:9092,192.168.2.141:9092,192.168.2.142:9092 --topic testkafka1 --from-beginning
hello
kafka
123
abc
1
2
3
4
5

2.2.1版本

在kafka2.8版本之前的版本，kakfa依赖zookeeper。此次测试使用kakfa 2.2.1版本。

安装包下载

直接在服务器用wget下载或者用迅雷下载好了上传也行。下载地址https://kafka.apache.org/downloads

[root@k8s-m1 ~]# wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz
1

单节点模式

zookeeper服务启动

Kafka 使用 ZooKeeper 如果你还没有ZooKeeper服务器，你需要先启动一个ZooKeeper服务器。 可以通过与kafka打包在一起的便捷脚本来快速简单地创建一个单节点ZooKeeper实例，当然也可单独下载zookeeper的安装包进行安装。

[root@k8s-m1 opt]# tar -xvf /root/kafka_2.12-2.2.1.tgz -C /tmp/
[root@k8s-m1 tmp]# cd /tmp/kafka_2.12-2.2.1/
[root@k8s-m1 kafka_2.12-2.2.1]# ./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties 
[root@k8s-m1 kafka_2.12-2.2.1]# jps
1573 QuorumPeerMain
1638 Jps
1
2
3
4
5
6

kafka服务启动和检查

[root@k8s-m1 kafka_2.12-2.2.1]# ./bin/kafka-server-start.sh -daemon config/server.properties
[root@k8s-m1 kafka_2.12-2.2.1]# jps
1573 QuorumPeerMain
4618 Kafka
4700 Jps
1
2
3
4
5

测试使用

#创建一个topic
[root@k8s-m1 kafka_2.12-2.2.1]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
Created topic test.
#查看topic列表
[root@k8s-m1 kafka_2.12-2.2.1]# ./bin/kafka-topics.sh --list --zookeeper localhost:2181
test
#查看topic详情
[root@k8s-m1 kafka_2.12-2.2.1]# ./bin/kafka-topics.sh --describe --zookeeper localhost:2181
Topic:test      PartitionCount:1        ReplicationFactor:1     Configs:
        Topic: test     Partition: 0    Leader: 0       Replicas: 0     Isr: 0
#发送消息
[root@k8s-m1 kafka_2.12-2.2.1]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>This is a message
>This is kafka    

#接收消息
[root@k8s-m1 kafka_2.12-2.2.1]#  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
This is a message
This is kafka
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

集群模式

zookeeper服务

为保证zookeeper的高可用，还是选择部署了3节点的zookeeper。
参考https://blog.csdn.net/margu_168/article/details/132598962，版本选择，可以通过查看解压后的kafka包中zookeeper的jar包。

[root@k8s-m1 libs]# ll /tmp/kafka_2.12-2.2.1/libs/zookeeper-3.4.13.jar
1

一般zookeeper后面的数字就代表可以使用的版本。

kafka集群部署

以下操作需要在3个节点都执行。注意在添加broker.id和advertised.listerners时每台服务器上的值不一样。

[root@k8s-m1 ~]# tar -xvf kafka_2.12-2.2.1.tgz -C /tmp/
[root@k8s-m1 ~]# mkdir /kafkalogs
[root@k8s-m1 ~]# sed -i '/^log.dirs=/d' /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 ~]# sed -i '/^broker.id/d' /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 ~]# sed -i '/^zookeeper.connect=/d' /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 ~]# sed -i '/^offsets.topic.replication.factor=1/d' /tmp/kafka_2.12-2.2.1/config/server.properties

[root@k8s-m1 kafka_2.12-2.2.1]# echo -e  "\nbroker.id=1" >> /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 kafka_2.12-2.2.1]# echo -e "zookeeper.connect=192.168.2.140:2181,192.168.2.141:2181,192.168.2.142:2181" >> /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 kafka_2.12-2.2.1]# echo -e "offsets.topic.replication.factor=3" >> /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 kafka_2.12-2.2.1]# echo -e  "advertised.listeners=PLAINTEXT://192.168.2.140:9092" >> /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 kafka_2.12-2.2.1]# echo -e  "log.dirs=/kafkalogs" >> /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 kafka_2.12-2.2.1]# echo -e  "auto.create.topics.enable=true" >>  /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 kafka_2.12-2.2.1]# echo -e  "delete.topic.enable=true" >>  /tmp/kafka_2.12-2.2.1/config/server.properties
#最后的效果如下，注意不同节点的区别
#k8s-m1
[root@k8s-m1 kafka_2.12-2.2.1]# vim config/server.properties 
......
broker.id=1
zookeeper.connect=192.168.2.140:2181,192.168.2.141:2181,192.168.2.142:2181
offsets.topic.replication.factor=3
advertised.listeners=PLAINTEXT://192.168.2.140:9092
log.dirs=/kafkalogs
auto.create.topics.enable=true
delete.topic.enable=true

#k8s-m2
[root@k8s-m2 ~]# vim /tmp/kafka_2.12-2.2.1/config/server.properties 
......
broker.id=2
zookeeper.connect=192.168.2.140:2181,192.168.2.141:2181,192.168.2.142:2181
offsets.topic.replication.factor=3
advertised.listeners=PLAINTEXT://192.168.2.141:9092
log.dirs=/kafkalogs
auto.create.topics.enable=true
delete.topic.enable=true

#k8s-m3
......
broker.id=3
zookeeper.connect=192.168.2.140:2181,192.168.2.141:2181,192.168.2.142:2181
offsets.topic.replication.factor=3
advertised.listeners=PLAINTEXT://192.168.2.142:9092
log.dirs=/kafkalogs
auto.create.topics.enable=true
delete.topic.enable=true
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

kafka服务启动和检查

三个节点依次启动kafka。检查结果中QuorumPeerMain为zookeeper的进程。一定要确保三个节点的进程都正常启动，如果不正常启动可以查看日志，目录为logs/server.log

[root@k8s-m1 kafka_2.12-2.2.1]# /tmp/kafka_2.12-2.2.1/bin/kafka-server-start.sh -daemon /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 kafka_2.12-2.2.1]# jps
15633 Jps
22170 QuorumPeerMain
13979 Kafka
1
2
3
4
5

测试使用

主题创建

#创建一个3副本，2分区的topic
[root@k8s-m1 kafka_2.12-2.2.1]#  bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 2 --topic my-replicated-topic
Created topic my-replicated-topic.
#查看创建好的topic
[root@k8s-m1 kafka_2.12-2.2.1]#  bin/kafka-topics.sh --list  --zookeeper localhost:2181  --topic my-replicated-topic
my-replicated-topic
#查看某个topic的具体情况
[root@k8s-m1 kafka_2.12-2.2.1]#  bin/kafka-topics.sh --describe   --zookeeper localhost:2181  --topic my-replicated-topic
Topic:my-replicated-topic       PartitionCount:2        ReplicationFactor:3     Configs:
        Topic: my-replicated-topic      Partition: 0    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3
        Topic: my-replicated-topic      Partition: 1    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
1
2
3
4
5
6
7
8
9
10
11

以下是对输出信息的解释。第一行给出了所有分区的摘要，下面的每行都给出了一个分区的信息。因为我们有两个分区，所以有两行。

• “leader”是负责给定分区所有读写操作的节点。每个节点都是随机选择的部分分区的领导者。
• “replicas”是复制分区日志的节点列表，不管这些节点是leader还是仅仅活着。
• “isr”是一组“同步”replicas，是replicas列表的子集，它活着并被指到leader。

生产信息

[root@k8s-m1 kafka_2.12-2.2.1]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic
>hello world     
>hello kafka  
>
1
2
3
4

消费信息

[root@k8s-m1 kafka_2.12-2.2.1]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic
hello world
hello kafka
1
2
3

容错性测试
从上面的describe可以看到，对于my-replicated-topic这个topic的分区0，它的leader是broker1，可以先将其杀死进行测试。

[root@k8s-m1 kafka_2.12-2.2.1]# jps
15633 Jps
22170 QuorumPeerMain
13979 Kafka
[root@k8s-m1 kafka_2.12-2.2.1]# kill -9 13979

#再次查看该topic
[root@k8s-m1 kafka_2.12-2.2.1]#  bin/kafka-topics.sh --describe  --zookeeper localhost:2181  --topic my-replicated-topic
Topic:my-replicated-topic       PartitionCount:2        ReplicationFactor:3     Configs:
        Topic: my-replicated-topic      Partition: 0    Leader: 2       Replicas: 1,2,3 Isr: 2,3
        Topic: my-replicated-topic      Partition: 1    Leader: 2       Replicas: 2,3,1 Isr: 2,3
[root@k8s-m1 kafka_2.12-2.2.1]# 
1
2
3
4
5
6
7
8
9
10
11
12

可以看到，对于my-replicated-topic这个topic的分区0，它的leader已经从broker1变成了broker2（Leader2）
不过，即便原先写入消息的leader已经不在，这些消息仍可用于消费，注意–bootstrap-server我们改成了的第二台服务器的IP，其实这个地方可以将3个IP地址全部写上。

[root@k8s-m1 kafka_2.12-2.2.1]# bin/kafka-console-consumer.sh --bootstrap-server 192.168.2.141:9092 --from-beginning --topic my-replicated-topichello world
hello world
hello kafka
1
2
3

使用Kafka Connect来导入/导出数据
创建测试数据

#先进之前停掉的broker节点启动起来
[root@k8s-m1 kafka_2.12-2.2.1]# /tmp/kafka_2.12-2.2.1/bin/kafka-server-start.sh -daemon /tmp/kafka_2.12-2.2.1/config/server.properties
[root@k8s-m1 kafka_2.12-2.2.1]# echo -e "foo\nbar" > test.txt
[root@k8s-m1 kafka_2.12-2.2.1]# bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties
1
2
3
4

这些包含在Kafka中的示例配置文件使用您之前启动的默认本地群集配置，并创建两个连接器： 第一个是源连接器，用于从输入文件读取行，并将其输入到 Kafka topic。 第二个是接收器连接器，它从Kafka topic中读取消息，并在输出文件中生成一行。

在启动过程中，你会看到一些日志消息，包括一些连接器正在实例化的指示。 一旦Kafka Connect进程启动，源连接器就开始从 test.txt 读取行并且 将它们生产到主题 connect-test 中，同时接收器连接器也开始从主题 connect-test 中读取消息， 并将它们写入文件 test.sink.txt 中。我们可以通过检查输出文件的内容来验证数据是否已通过整个pipeline进行交付。

#查看，注意路径
[root@k8s-m1 kafka_2.12-2.2.1]# more test.sink.txt
foo
bar
1
2
3
4

大家可以自行查看创建连接器过程中使用的配置文件，里面有定义各输入输出文件的名字。

注意，导入的数据存储在Kafka topic connect-test 中，因此我们也可以运行一个console consumer（控制台消费者）来查看 topic 中的数据（或使用custom consumer（自定义消费者）代码进行处理）：

[root@k8s-m1 kafka_2.12-2.2.1]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}
1
2
3

而如果没有将连接器断开，连接器将一直处理数据，所以我们可以将数据添加到文件中，并看到它在pipeline 中移动

[root@k8s-m1 kafka_2.12-2.2.1]# echo "hello world" >> test.txt
1

我们可以看到这一行出现在控制台用户输出和接收器文件中。

[root@k8s-m1 kafka_2.12-2.2.1]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}
{"schema":{"type":"string","optional":false},"payload":"hello world"}
1
2
3
4

以系统服务方式启动kafka

新建文件/usr/lib/systemd/system/kafka.service，并编写以下内容，注意使用的路径要与自己实际环境中相匹配。

[root@k8s-m1 bin]# vim  /usr/lib/systemd/system/kafka.service 
[Unit]
Requires=zookeeper.service
After=zookeeper.service
[Service]
Type=simple
LimitNOFILE=1048576
ExecStart=/tmp/kafka_2.12-2.2.1/bin/kafka-server-start.sh /tmp/kafka_2.12-2.2.1/config/server.properties
ExecStop=/tmp/kafka_2.12-2.2.1/bin/kafka-server-stop.sh
Restart=Always
[Install]
WantedBy=multi-user.target
1
2
3
4
5
6
7
8
9
10
11
12

重要
在 kafka的这个kafka-run-class.sh 脚本中，由于默认使用的java为/usr/bin/java，而我们安装的java不在这个路径下。如果不作配置，kafka是不能正常启动的，会报类似tmp/kafka_2.12-2.2.1/bin/kafka-run-class.sh: line 306: exec: java: not found的错误，解决方法有二。一是建立一个软连接，方法如下：

#查看我们实际环境中Java的路径，并创建软连接
[root@k8s-m1 bin]# echo $JAVA_HOME
/opt/jdk1.8.0_65
[root@k8s-m1 bin]# ln -s /opt/jdk1.8.0_65/bin/java /usr/bin/java
1
2
3
4

方法二，编辑kafka-run-class.sh 这个文件，在最前面添加JAVA_HOME的实际路径。

[root@k8s-m1 bin]# vim kafka-run-class.sh 
# limitations under the License.
JAVA_HOME='/opt/jdk1.8.0_65'

if [ $# -lt 1 ];
then
  echo "USAGE: $0 [-daemon] [-name servicename] [-loggc] classname [opts]"
  exit 1
fi
1
2
3
4
5
6
7
8
9

以上两种方法任选其一即可。

#以systemctl的方式启动kafka
[root@k8s-m1 bin]# systemctl start  kafka.service 
[root@k8s-m1 bin]# jps
22170 QuorumPeerMain
32363 Kafka
7582 Jps
1
2
3
4
5
6

更多关于kafka的知识分享，请前往博客主页。编写过程中，难免出现差错，敬请指出