【Kafka】|?總結/Edison Zhou

1準備工作

這里我們使用一臺Linux CentOS系統的服務器來模擬三個Kafka Broker的偽集群（即一臺server上開三個不同端口）環境用于學習測試，大概的準備工作有兩個：

安裝Docker

# wget https://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo -O
/etc/yum.repos.d/docker-ce.repo
# yum -y install docker
# systemctl enable docker && systemctl start docker
# docker --version

安裝Docker Compose

# sudo curl -L "https://github.com/docker/compose/releases/download/1.28.2/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
# sudo chmod +x /usr/local/bin/docker-compose
# docker-compose --version

2部署Kafka環境

準備docker-compose.yml文件

這里我的宿主機IP是172.16.16.4，你需要改為你自己的。

具體的 docker-compose.yml 文件內容如下：

version: '3.8'
services:zookeeper:image: wurstmeister/zookeepercontainer_name: zookeeperports:- "2181:2181"restart: alwayskafka1:image: wurstmeister/kafkadepends_on: [ zookeeper ]container_name: kafka1ports:- "9091:9091"environment:HOSTNAME: kafka1KAFKA_BROKER_ID: 0KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9091KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9091KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181/kafkaextra_hosts:kafka1: 172.16.16.4kafka2:image: wurstmeister/kafkadepends_on: [ zookeeper ]container_name: kafka2ports:- "9092:9092"environment:HOSTNAME: kafka2KAFKA_BROKER_ID: 1KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:9092KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181/kafkaextra_hosts:kafka2: 172.16.16.4kafka3:image: wurstmeister/kafkadepends_on: [ zookeeper ]container_name: kafka3ports:- "9093:9093"environment:HOSTNAME: kafka3KAFKA_BROKER_ID: 2KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka3:9093KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9093KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181/kafkaextra_hosts:kafka3: 172.16.16.4

部署Zookeeper和Kafka

將docker-compose.yml文件拷貝到服務器，緊接著在該文件目錄下執行：docker-compose up -d 即可快速完成部署。

# docker-compose up -d
Building with native build. Learn about native build in Compose here: https://docs.docker.com/go/compose-native-build/
Creating network "kafka_default" with the default driver
Creating zookeeper ... done
Creating kafka1    ... done
Creating kafka3    ... done
Creating kafka2    ... done

部署完成后，通過執行：docker-compose ps 來驗證一下。

# docker-compose psName                 Command               State                                  Ports                                
-------------------------------------------------------------------------------------------------------------------------
kafka1      start-kafka.sh                   Up      0.0.0.0:9091->9091/tcp,:::9091->9091/tcp                            
kafka2      start-kafka.sh                   Up      0.0.0.0:9092->9092/tcp,:::9092->9092/tcp                            
kafka3      start-kafka.sh                   Up      0.0.0.0:9093->9093/tcp,:::9093->9093/tcp                            
zookeeper   /bin/sh -c /usr/sbin/sshd  ...   Up      0.0.0.0:2181->2181/tcp,:::2181->2181/tcp, 22/tcp, 2888/tcp, 3888/tcp

其中，docker-compose.yml中HOSTNAME和extra_hosts的結合使用，會在容器中的/etc/hosts中增加一條記錄，通過執行：docker exec -it kafka1 cat /etc/hosts 來驗證一下。

# docker exec -it kafka1 cat /etc/hosts
.....
172.16.16.4  kafka1
.....

3容器內驗證與測試

安裝部署完成后，首先，我們進入容器內部，來驗證和測試一下。

docker exec -it kafka1 bash

創建測試Topic

創建一個測試用的topic：testtopic，此topic配置了2個分區，無額外的副本。

kafka-topics.sh --create --zookeeper 172.16.16.4:2181/kafka --replication-factor 1 --partitions 2 --topic testtopic

模擬Producer

重新打開一個窗口，進入容器內部，模擬一個producer，在控制臺隨意發送一些字符串消息。

kafka-console-producer.sh --topic=testtopic --broker-list kafka1:9091,kafka2:9092,kafka3:9093
>tests^H
>haha

模擬Consumer

重新打開一個窗口，進入容器內部，模擬一個consumer，設置從頭開始消費，會收到producer發來的字符串消息。

kafka-console-consumer.sh --bootstrap-server kafka1:9091,kafka2:9092,kafka3:9093 --from-beginning --topic testtopic
tests
haha

可以看到，consumer成功接收到了producer發來的兩個string類型的message。

4使用GUI工具：Kafka Tool

Kafka Tool是一個用于管理和使用Apache Kafka集群的GUI應用程序。Kafka Tool提供了一個較為直觀的UI可讓用戶快速查看Kafka集群中的對象以及存儲在topic中的消息，提供了一些專門面向開發人員和管理員的功能。

NOTE：目前Kafka Tool已改名為Offset Explorer，不過我還是傾向于叫它 Kafka Tool。

下載地址：https://www.kafkatool.com/download.html

使用方式比較簡單，首先創建一個Connection，需要填寫的配置如下：

NOTE：這里bootstrap-servers參數列表填寫的是主機名，你可以先在自己電腦上的hosts文件中添加這個映射

然后點擊Test測試無誤就進入了以下界面，可以看到我們剛剛創建的topic有兩個分區已經可以顯示出來了。

然后更改一下顯示內容的類型：從Byte Array改為String，方便查看。

通過上面的producer.sh再發送一些消息，然后通過kafka tool來查看一下消息：

End總結

本文總結了Kafka的測試環境搭建過程，本文選擇的是基于Docker來搭建非宿主機直接搭建，加之官方并沒有推出官方的Docker鏡像，因此建議生產環境還是不要通過Docker來搭建，開發/測試環境是可以的，快速且高效。

參考資料

極客時間，胡夕《Kafka核心技術與實戰》

B站，尚硅谷《Kafka 3.x入門到精通教程》

年終總結：Edison的2020年終總結

數字化轉型：我在傳統企業做數字化轉型

C#刷題：C#刷劍指Offer算法題系列文章目錄

.NET面試：.NET開發面試知識體系

.NET大會：2020年中國.NET開發者大會PDF資料