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

    概述

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

    服务器环境规划

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

    IPhostname
    192.168.2.140k8s-m1
    192.168.2.141k8s-m2
    192.168.2.142k8s-m3

    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的知识分享，请前往博客主页。编写过程中，难免出现差错，敬请指出

  • 相关阅读:
    OpenHarmony鸿蒙南向开发案例：【智能猫眼（基于3518开发板）】
    AJAX学习（四）
    人人都会数据分析
    victoriaMetrics无法获取抓取target的问题
    APP逆向案例之（三）sign 参数破解
    13、用户web层服务（一）
    Java ByteArrayOutputStream.toString()方法具有什么功能呢?
    视频转换器WinX HD Video Converter mac中文特点介绍
    Python pyenv install 下载安装慢（失败）完美解决
    Kafka消费者订阅指定主题(subscribe)或分区(assign)详解
  • 原文地址：https://blog.csdn.net/margu_168/article/details/132736406