kafka3.X集群安装(不使用zookeeper)

た入场券 2024-02-21 12:26 64阅读 0赞

### 一、[kafka集群][kafka]实例角色规划 ###

在本专栏的之前的一篇文章《kafka3种zk的替代方案》已经为大家介绍过在kafka3.0种已经可以将zookeeper去掉。

![029569adb193f9b7a109935f7f8fc79e.png][]

上图中黑色代表broker（消息[代理服务][Link 1]），褐色/蓝色代表Controller(集群控制器服务)

>  *  左图（kafka2.0）：一个集群所有节点都是broker角色，kafka从三个broker中选举出来一个Controller控制器，控制器将集群元数据信息（比如主题分类、消费进度等）保存到zookeeper，用于集群各节点之间分布式交互。
>  *  右图（kafka3.0）：假设一个集群有四个broker，指定三个作为Conreoller角色(蓝色)，从三个Controller中选举出来一个Controller作为主控制器（褐色），其他的2个备用。zookeeper不再被需要！相关的元数据信息以kafka日志的形式存在（即：以消息队列消息的形式存在）。
>  *  controller通信端口：9093, 作用与zk的2181端口类似 。

在搭建kafka3.0集群之前, 我们需要先做好[kafka][kafka 1]实例角色规划。(四个broker, 需要通过主动配置指定三个作为Controller, Controller需要奇数个, 这一点和zk是一样的)

<table> 
 <thead> 
  <tr> 
   <th>主机名称</th> 
   <th>ip</th> 
   <th>角色</th> 
   <th><code>node.id</code></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>kafka-vm1</td> 
   <td>192.168.1.111</td> 
   <td>broker,controller</td> 
   <td>1</td> 
  </tr> 
  <tr> 
   <td>kafka-vm2</td> 
   <td>192.168.1.112</td> 
   <td>broker,controller</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td>kafka-vm3</td> 
   <td>192.168.1.113</td> 
   <td>broker,controller</td> 
   <td>3</td> 
  </tr> 
  <tr> 
   <td>kafka-vm4</td> 
   <td>192.168.1.114</td> 
   <td>broker</td> 
   <td>4</td> 
  </tr> 
 </tbody> 
</table>

### 二、准备工作 ###

>  *  kafka3.x不再支持JDK8，建议安装JDK11或JDK17。
>  *  新建kafka持久化日志数据mkdir -p /data/kafka;并保证安装kafka的用户具有该目录的读写权限。

**各个机器节点执行：**

# 安装jdk(kafka3.x不再支持JDK8，建议安装JDK11或JDK17, 这里安装jdk11)
    # 下载安装jdk11, 参考: https://blog.csdn.net/justlpf/article/details/127268046
     
    # 下载kafka
    adduser kafka
    cd /opt
    wget https://downloads.apache.org/kafka/3.3.1/kafka_2.12-3.3.1.tgz
    tar -xf kafka_2.12-3.3.1.tgz
     
    chown -R kafka:kafka kafka_2.12-3.3.1*
     
    mkdir -p /data/kafka
    chown -R kafka:kafka /data/kafka

**vi /etc/hosts，各个节点，添加如下内容：**

192.168.1.111 data-vm1
    192.168.1.112 data-vm2
    192.168.1.113 data-vm3
    192.168.1.114 data-vm4

### 三、修改Kraft协议配置文件 ###

在kafka3.x版本中，使用Kraft协议代替zookeeper进行集群的Controller选举，所以要针对它进行配置。

vi /opt/kafka_2.12-3.3.1/config/kraft/server.properties

**具体配置参数如下：**

# data-vm1节点
    node.id=1
    process.roles=broker,controller
    listeners=PLAINTEXT://data-vm1:9092,CONTROLLER://data-vm1:9093
    advertised.listeners=PLAINTEXT://:9092
    controller.quorum.voters=1@data-vm1:9093,2@data-vm2:9093,3@data-vm3:9093
    log.dirs=/data/kafka/
     
    # data-vm2节点
    node.id=2
    process.roles=broker,controller
    listeners=PLAINTEXT://data-vm2:9092,CONTROLLER://data-vm2:9093
    advertised.listeners=PLAINTEXT://:9092
    controller.quorum.voters=1@data-vm1:9093,2@data-vm2:9093,3@data-vm3:9093
    log.dirs=/data/kafka/
     
    # data-vm3节点
    node.id=3
    process.roles=broker,controller
    listeners=PLAINTEXT://data-vm3:9092,CONTROLLER://data-vm3:9093
    advertised.listeners=PLAINTEXT://:9092
    controller.quorum.voters=1@data-vm1:9093,2@data-vm2:9093,3@data-vm3:9093
    log.dirs=/data/kafka/

>  *  `node.id``：这将作为集群中的节点 ID,唯一标识，按照我们事先规划好的（上文），在不同的服务器上这个值不同。其实就是kafka2.0中的broker.id,只是在3.0版本中kafka实例不再只担任broker角色，也有可能是controller角色，所以改名叫做node节点。`
>  *  `process.roles``：一个节点可以充当broker或controller或两者兼而有之。按照我们事先规划好的（上文），在不同的服务器上这个值不同。多个角色用逗号分开。`
>  *  `listeners``: broker将使用9092端口，而kraft controller控制器将使用9093端口。`
>  *  `advertised.listeners``: 这里指定kafka通过代理暴漏的地址，如果都是局域网使用，就配置PLAINTEXT://:9092即可。`
>  *  `controller.quorum.voters``：这个配置用于指定controller主控选举的投票节点，所有process.roles包含controller角色的规划节点都要参与，即：zimug1、zimug2、zimug3。其配置格式为:node.id1@host1:9093,node.id2@host2:9093`
>  *  `log.dirs``：kafka 将存储数据的日志目录，在准备工作中创建好的目录。`

所有kafka节点都要按照上文中的节点规划进行配置，完成config/kraft/server.properties配置文件的修改。

### 四、格式化存储目录 ###

生成一个唯一的集群ID（在一台kafka服务器上执行一次即可），这一个步骤是在安装kafka2.0版本的时候不存在的。

$ /opt/kafka_2.12-3.3.1/bin/kafka-storage.sh random-uuid
    SzIhECn-QbCLzIuNxk1A2A

> 使用生成的集群ID+配置文件格式化存储目录log.dirs,
> 
> 所以这一步确认配置及路径确实存在，
> 
> 并且kafka用户有访问权限（检查准备工作是否做对）。
> 
> **每一台主机服务器都要执行命令:**

/opt/kafka_2.12-3.3.1/bin/kafka-storage.sh format \
    -t SzIhECn-QbCLzIuNxk1A2A \
    -c /opt/kafka_2.12-3.3.1/config/kraft/server.properties

格式化操作完成之后，log.dirs目录下多出一个Meta.properties文件，存储了当前的kafka节点的id(node.id)，当前节点属于哪个集群(cluster.id)

[root@data-vm2 ~]# ll /data/kafka/
    总用量 8
    -rw-r--r--. 1 root root 249 10月 11 18:23 bootstrap.checkpoint
    -rw-r--r--. 1 root root  86 10月 11 18:23 meta.properties
     
    $ cat /data/kafka/meta.properties
    #
    #Tue Apr 12 07:39:07 CST 2022
    node.id=1
    version=1
    cluster.id=SzIhECn-QbCLzIuNxk1A2A

### 五、 启动集群，完成基础测试 ###

> zimug1 zimug2 zimug3是三台应用服务器的主机名称(参考上文中的角色规划)，实现方式已经在本专栏《linux主机与ip解析》中进行了说明。将下面的命令集合保存为一个shell脚本，并赋予执行权限。执行该脚本即可启动kafka集群所有的节点，前提是：你已经按照本专栏的《集群各节点之间的ssh免密登录》安装方式做了集群各节点之间的ssh免密登录。

**启动命令：**

bin/kafka-server-start.sh \
    /opt/kafka_2.12-3.3.1/config/kraft/server.properties
     
    # 后台运行
    nohup bin/kafka-server-start.sh \
    /opt/kafka_2.12-3.3.1/config/kraft/server.properties 2>&1 &

**脚本：**

#!/bin/bash
    kafkaServers='data-vm1 data-vm2 data-vm3'
    #启动所有的kafka
    for kafka in $kafkaServers
    do
        ssh -T $kafka <<EOF
        nohup /opt/kafka_2.12-3.3.1/bin/kafka-server-start.sh /opt/kafka_2.12-3.3.1/config/kraft/server.properties 1>/dev/null 2>&1 &
    EOF
    echo 从节点 $kafka 启动kafka3.0...[ done ]
    sleep 5
    done

### 六、一键停止集群脚本 ###

一键停止kafka集群各节点的脚本，与启动脚本的使用方式及原理是一样的。

**停止命令：**

/opt/kafka_2.12-3.3.1/bin/kafka-server-stop.sh

**执行脚本：**

#!/bin/bash
    kafkaServers='data-vm1 data-vm2 data-vm3'
    #停止所有的kafka
    for kafka in $kafkaServers
    do
        ssh -T $kafka <<EOF
        cd /opt/kafka_2.12-3.3.1
        bin/kafka-server-stop.sh
    EOF
    echo 从节点 $kafka 停止kafka...[ done ]
    sleep 5
    done

### 七、测试Kafka集群 ###

#### 7.1 创建topic ####

[root@data-vm1 kafka_2.12-3.3.1]# bin/kafka-topics.sh \
    --create \
    --topic quickstart-events \
    --bootstrap-server data-vm4:9092
     
    Created topic quickstart-events.
    [root@data-vm1 kafka_2.12-3.3.1]#
     
    # 
    [root@data-vm1 kafka_2.12-3.3.1]# bin/kafka-topics.sh \
    --create \
    --topic quickstart-events \
    --bootstrap-server data-vm1:9092,data-vm2:9092,data-vm3:9092

#### 7.2 查看topic列表 ####

bin/kafka-topics.sh \
    --list \
    --bootstrap-server data-vm4:9092
     
    # 
    bin/kafka-topics.sh \
    --list \
    --bootstrap-server data-vm1:9092,data-vm2:9092,data-vm3:9092,data-vm4:9092

#### 7.3 查看消息详情 ####

[root@data-vm1 kafka_2.12-3.3.1]# bin/kafka-topics.sh \
    --describe \
    --topic quickstart-events \
    --bootstrap-server data-vm3:9092
     
    Topic: quickstart-events        TopicId: zSOJC6wNRRGQ4MudfHLGvQ PartitionCount: 1       ReplicationFactor: 1    Configs: segment.bytes=1073741824
            Topic: quickstart-events        Partition: 0    Leader: 1       Replicas: 1     Isr: 1
     
    [root@data-vm1 kafka_2.12-3.3.1]#

#### 7.4 生产消息 ####

[root@data-vm1 kafka_2.12-3.3.1]# bin/kafka-console-producer.sh \
    --topic quickstart-events \
    --bootstrap-server data-vm1:9092
     
    # 参考: 创建并配置topic
    bin/kafka-topics.sh \
    --bootstrap-server localhost:9092 \
    --create \
    --topic my-topic \
    --partitions 1 \
    --replication-factor 1 \
    --config max.message.bytes=64000 \
    --config flush.messages=1
     
    # ------------------------- 参考 ------------------------ #
    # 1: 修改已创建topic配置
    # (Overrides can also be changed or set later using the alter configs command.)
    bin/kafka-configs.sh \
    --bootstrap-server localhost:9092 \
    --entity-type topics \
    --entity-name my-topic \
    --alter \
    --add-config max.message.bytes=128000
     
    # 2: 检查已修改的topic配置是否生效
    # (To check overrides set on the topic you can do)
    bin/kafka-configs.sh \
    --bootstrap-server localhost:9092 \
    --entity-type topics \
    --entity-name my-topic \
    --describe
     
    # 3. 恢复到原来的配置
    # (To remove an override you can do)
    bin/kafka-configs.sh \
    --bootstrap-server localhost:9092 \
    --entity-type topics \
    --entity-name my-topic \
    --alter \
    --delete-config max.message.bytes
     
    # 4. 增加分区数
    # (To add partitions you can do)
    bin/kafka-topics.sh \
    --bootstrap-server broker_host:port \
    --alter \
    --topic my_topic_name \
    --partitions 40
     
    # 5. 添加配置
    # (To add configs:)
    bin/kafka-configs.sh \
    --bootstrap-server broker_host:port \
    --entity-type topics \
    --entity-name my_topic_name \
    --alter \
    --add-config x=y
     
    # 6. 移除配置
    # (To remove a config:)
    bin/kafka-configs.sh \
    --bootstrap-server broker_host:port \
    --entity-type topics \
    --entity-name my_topic_name \
    --alter \
    --delete-config x
     
    # 7. 删除topic
    # (And finally deleting a topic:)
    bin/kafka-topics.sh \
    --bootstrap-server broker_host:port \
    --delete \
    --topic my_topic_name

#### 7.5 消费消息 ####

bin/kafka-console-consumer.sh \
    --topic quickstart-events \
    --from-beginning \
    --bootstrap-server data-vm4:9092

#### 7.6 查看消费者组 ####

# 检查消费者postition
    # Checking consumer position
    bin/kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --describe \
    --group my-group
     
      TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
      my-topic                       0          2               4               2          consumer-1-029af89c-873c-4751-a720-cefd41a669d6   /127.0.0.1                     consumer-1
      my-topic                       1          2               3               1          consumer-1-029af89c-873c-4751-a720-cefd41a669d6   /127.0.0.1                     consumer-1
      my-topic                       2          2               3               1          consumer-2-42c1abd4-e3b2-425d-a8bb-e1ea49b29bb2   /127.0.0.1                     consumer-2

#### 7.7 查看消费者组列表 ####

# list all consumer groups across all topics
    bin/kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --list
     
      test-consumer-group
     
     
    # To view offsets, as mentioned earlier, 
    # we "describe" the consumer group like this:
    bin/kafka-consumer-groups.sh \
    --bootstrap-server localhost:9092 \
    --describe \
    --group my-group
     
      TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                    HOST            CLIENT-ID
      topic3          0          241019          395308          154289          consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1      consumer2
      topic2          1          520678          803288          282610          consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1      consumer2
      topic3          1          241018          398817          157799          consumer2-e76ea8c3-5d30-4299-9005-47eb41f3d3c4 /127.0.0.1      consumer2
      topic1          0          854144          855809          1665            consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1      consumer1
      topic2          0          460537          803290          342753          consumer1-3fc8d6f1-581a-4472-bdf3-3515b4aee8c1 /127.0.0.1      consumer1
      topic3          2          243655          398812          155157          consumer4-117fe4d3-c6c1-4178-8ee9-eb4a3954bee0 /127.0.0.1      consumer4
     
    # 更多配置参考:
    # https://kafka.apache.org/32/documentation.html#uses

[kafka]: https://so.csdn.net/so/search?q=kafka%E9%9B%86%E7%BE%A4&spm=1001.2101.3001.7020
[029569adb193f9b7a109935f7f8fc79e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/21/9ff87d028b4d4e2a85d52fa396ae710c.png
[Link 1]: https://so.csdn.net/so/search?q=%E4%BB%A3%E7%90%86%E6%9C%8D%E5%8A%A1&spm=1001.2101.3001.7020
[kafka 1]: https://so.csdn.net/so/search?q=kafka&spm=1001.2101.3001.7020