【Kafka集群搭建，包括docker整合kafka、zookeeper、nginx部署】

短命女 2024-03-16 17:10 7阅读 0赞

## Kafka集群搭建 ##

### //启动zookeeper: ###

bin/zookeeper-server-start.sh config/zookeeper.properties

### //启动集群中所有的kafka ###

bin/kafka-server-start.sh -daemon config/server.properties

### //创建topic，设置副本和分区： ###

bin/kafka-topics.sh --create --zookeeper 192.168.1.5:2181 --replication-factor 4 --partitions 4 --topic test

–replication-factor 4（设置副本为4）–partitions 4（设置分区为4\{最优为4\}）

### //删除topic ###

./bin/kafka-topics.sh --delete --zookeeper 192.168.1.5:2181 --topic topic名称

#### 方法一：手动删除法，通过zookeeper客户端删除topic（Delete topic时显示“Topic test is marked for deletion”） ####

需要进入zookeeper内部进行删除

./zookeeper-shell.sh 192.168.1.5:2181
    ls /brokers/topics
    rmr /brokers/topics/test
    quit

#### 方法二：快速配置删除法（简单粗暴，如果这个主题有程序还在消费者，此时KAFKA就game over） ####

1.kafka启动之前，在server.properties配置：**delete.topic.enable=true**  
2.执行命令bin/kafka-topics.sh –delete –topic test –zookeeper zk:2181或者使用kafka-manager集群管理工具删除  
注意：如果kafka启动之前没有配置delete.topic.enable=true，topic只会标记为marked for deletion，加上配置，重启kafka，之前的topick就真正删除了。

### //发送消息 ###

bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test（windows版本）
    bin/kafka-console-producer.sh --broker-list 192.168.1.5:9092 --topic test

### //启动consumer接受消息 ###

bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.5:9092 --topic test --from-beginning

### //启动多个kafka（新的节点） ###

bin/kafka-server-start.sh config/server-1.properties &
    bin/kafka-server-start.sh config/server-2.properties &

### //扩容分区 ###

\#把liNingShortSleeveTopic主题扩容为6个分区。  
\#注意：目前不支持减少分区，扩容前必须存在这个主题。

bin/kafka-topics.sh  --zookeeper 192.168.1.5:2181 --alter --topic test --partitions 4

### //查询某个Topic的详细信息（–describe） ###

bin/kafka-topics.sh --describe --zookeeper 192.168.1.5:2181 --topic my-replicated-topic
    bin/kafka-topics.sh --describe --zookeeper 192.168.1.5:2181 --topic test,track,user

![\[图片\]][ef24576236794d7c8e06d1ab3b0a2d96.png]

#### 名词说明（如partition） ####

##### Partition: #####

kafka 中 Topic 被分成多个 Partition 分区。Topic 是一个逻辑概念，Partition 是最小的存储单元，掌握着一个 Topic 的部分数据。每个 Partition 都是一个单独的 log 文件，每条记录都以追加的形式写入。

> 1、 一个Topic可以配置几个Partition，  
> 2、 Producer发送的消息分到不同的Partition中  
> 3、Consumer接收数据是按照 Consumer Group（消费组）来接接收，kafka确保每个Partition在同一个Consumer Group（消费组）只能由其中一个消费者消费。

##### Partitioner（分区器）: #####

分区选择器，作用就是为消息分配分区。默认使用murmur2对key进行hash计算，然后对于总分区数去模，得到对应的分区号，然后消息就发往该分区。Producer可以实现自定义分区器，或者指定特定的分区。只需要实现Partitioner接口。

##### Partition Leader #####

Kafka为了到达高可用，每个分区(Partition)都会多个副本，其中有且仅有一个作为leader,leader负责当前消息的读写Partition。

##### Partition Follower #####

所有 Follower 都需要从 Leader 同步消息，Follower 与 Leader 始终保持消息同步。  
partition leader 与 follower 是主备关系，而非主从。

##### AR #####

Assigned Replicas 同步副本，AR=ISR+OSR

##### ISR #####

ISR，In-Sync Replicas，是指及时副本同步列表，消息同步的频率或者时效达到一定要求副本。

##### OSR #####

Outof-Sync Replicas 消息同步存在滞后情况的副本集合

### //查看某个kafka所有的topic列表 ###

bin/kafka-topics.sh --list --zookeeper 192.168.1.5:2181

![\[图片\]][0814379b2fdf4f79ae890efca35ec613.png]

### //查看kafka某个topic中的数据量 ###

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.1.5:9092 --topic testpro --time -1

![\[图片\]][2c5f87cf47f749c2b11d7af25927790e.png]

### //查看某个用户组的kafka消息消费情况，有没有数据积压 ###

bin/kafka-consumer-groups.sh --bootstrap-server 192.168.1.5:9092 --describe --group group-1

![\[图片\]][b6be6c1893f44db3b8aa7ef408616952.png]

### //查看topic某分区数据偏移量(offset)最大值，就是看目前Kafka里有多少条消息 ###

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 192.168.1.5:9092 --topic topic队列名字 --partitions 0

![\[图片\]][d805ca07fee34105ab01887bf9d663a9.png]

## docker整合zookeeper: ##

如果没有docker会先下载docker(**选第三个**)

docker run -d --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 wurstmeister/zookeeper

### 进入docker容器 ###

docker exec -it kafka3 /bin/bash

### docker容器中没有vi编辑器的解决办法 ###

mv /etc/apt/sources.list /etc/apt/sources.list.bak
    
    cat <<EOF >/etc/apt/sources.list
    deb http://mirrors.ustc.edu.cn/debian stable main contrib non-free
    deb http://mirrors.ustc.edu.cn/debian stable-updates main contrib non-free
    EOF
    
    apt update
    
    apt install vim
    
    mv /etc/apt/sources.list.bak /etc/apt/sources.list

## docker创建启动kafka： ##

> docker run -it -d --name 要取的别名 -p 宿主机端口:容器端口 -v 宿主机文件存储位置:容器内文件位置  
> 镜像名:Tag /bin/bash

> \-it 表示 与容器进行交互式启动  
> \-d 表示可后台运行容器 （守护式运行）  
> –name 给要运行的容器 起的名字 /bin/bash 交互路径  
> \-p 将容器的端口映射到宿主机上，通过宿主机访问内部端口  
> \-v 将容器内的指定文件夹挂载到宿主机对应位置

docker run --name kafka1 \
       -p 9092:9092 \
       -e KAFKA_BROKER_ID=1 \
       -e KAFKA_ZOOKEEPER_CONNECT=192.168.1.5:2181 \
       -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.1.5:9092 \
       -e KAFKA_LISTENERS=PLAINTEXT://:9092 \
       -d wurstmeister/kafka

docker run --name kafka2 \
       -p 9093:9093 \
       -e KAFKA_BROKER_ID=2 \
       -e KAFKA_ZOOKEEPER_CONNECT=192.168.1.5:2181 \
       -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.1.5:9093 \
       -e KAFKA_LISTENERS=PLAINTEXT://:9093 \
       -d wurstmeister/kafka

docker run --name kafka3 \
       -p 19092:19092 \
       -e KAFKA_BROKER_ID=3 \
       -e KAFKA_ZOOKEEPER_CONNECT=192.168.1.5:2181 \
       -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.1.6:19092 \
       -e KAFKA_LISTENERS=PLAINTEXT://:19092 \
       -d wurstmeister/kafka

docker run --name kafka4 \
       -p 19093:19093 \
       -e KAFKA_BROKER_ID=4 \
       -e KAFKA_ZOOKEEPER_CONNECT=192.168.1.5:2181 \
       -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.1.6:19093 \
       -e KAFKA_LISTENERS=PLAINTEXT://:19093 \
       -d wurstmeister/kafka

## 进入kafka本地的zookeeper ##

## 进入zookeeper客户端 ##

[root@localhost bin] ./zookeeper-shell.sh 192.168.1.5:2181
    (提示下面的表示进入成功)
    Connecting to localhost:2181
    Welcome to ZooKeeper!
    JLine support is disabled
    
    WATCHER::
    
    WatchedEvent state:SyncConnected type:None path:null

## 查看客户端里面的节点信息 ##

ls /
    [admin, brokers, cluster, config, consumers, controller, controller_epoch, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper]
    ls /brokers/ids
    [0, 1, 2]

## nginx整合kafka: ##

nginx版本 1.12.2

### 1.将kafka的c客户端源码clone到本地 ###

git clone https://github.com/edenhill/librdkafka

### 2.进行编译 需要有gcc gcc-c++ pcre-devel zlib-devel环境 ###

yum install -y gcc gcc-c++ pcre-devel zlib-devel
    ./configure
    make && make install

### 3.安装nginx整合kafka的插件 ###

git clone https://github.com/brg-liuwei/ngx_kafka_module

### 4.进入nginx源码包 将插件进行编译 ###

./configure --add-module=/usr/local/src/ngx_kafka_module/
      make
      make install

这里遇到的问题,编译的时候会报错 这里把报错的源码位置注释一下就好了  
并且按照网上给的提示，将objs里面的Makefile文件中的编译命令 WErroe 删除

启动nginx，报错，找不到kafka.so.1的文件  
加载so库

> echo “/usr/local/lib” >> /etc/ld.so.conf ldconfig

### 5.修改nginx配置文件,启动nginx ###

http{
        
    kafka;
    kafka_broker_list host1 host2;
        server{
        
            location = /kafka/track {
        
                    kafka_topic track;
            }
    
            location = /kafka/user {
        
                    kafka_topic user;
            }
        }
    }

### 测试nginx整合kafka ###

curl localhost/kafka/track -d "message send to kafka topic"

## 记住，修改完 Kafka 的配置文件后一定要重启 Kafka 才能生效 ##

![在这里插入图片描述][4d9a7408c5774e85be99c1b6f43bf950.png]

[ef24576236794d7c8e06d1ab3b0a2d96.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/fe4b83c48c084b84939dd829797211cc.png
[0814379b2fdf4f79ae890efca35ec613.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/583ecaab8eac44059924e1403a67da5f.png
[2c5f87cf47f749c2b11d7af25927790e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/8a7a07651b9543448962b4b2110ef423.png
[b6be6c1893f44db3b8aa7ef408616952.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/b95195f361d5412c98de6abb69901da1.png
[d805ca07fee34105ab01887bf9d663a9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/b71ada53d5f545529fbda5644222304b.png
[4d9a7408c5774e85be99c1b6f43bf950.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/72cd721035104f95a1e376e701d24e99.png