一、前言

1、Kafka簡介

Kafka是一個開源的分布式消息引擎/消息中間件，同時Kafka也是一個流處理平臺。Kakfa支持以發布/訂閱的方式在應用間傳遞消息，同時并基于消息功能添加了Kafka Connect、Kafka Streams以支持連接其他系統的數據(Elasticsearch、Hadoop等)

Kafka最核心的最成熟的還是他的消息引擎，所以Kafka大部分應用場景還是用來作為消息隊列削峰平谷。另外，Kafka也是目前性能最好的消息中間件。

2、Kafka架構

在Kafka集群(Cluster)中，一個Kafka節點就是一個Broker，消息由Topic來承載，可以存儲在1個或多個Partition中。發布消息的應用為Producer、消費消息的應用為Consumer，多個Consumer可以促成Consumer Group共同消費一個Topic中的消息。

概念/對象簡單說明

Broker	Kafka節點
Topic	主題，用來承載消息
Partition	分區，用于主題分片存儲
Producer	生產者，向主題發布消息的應用
Consumer	消費者，從主題訂閱消息的應用
Consumer Group	消費者組，由多個消費者組成

3、準備工作

1、Kafka服務器

準備3臺CentOS服務器，并配置好靜態IP、主機名

服務器名IP說明

kafka01	192.168.88.51	Kafka節點1
kafka02	192.168.88.52	Kafka節點2
kafka03	192.168.88.53	Kafka節點3

軟件版本說明

項說明

Linux Server	CentOS 7
Kafka	2.3.0

2、ZooKeeper集群

Kakfa集群需要依賴ZooKeeper存儲Broker、Topic等信息，這里我們部署三臺ZK

服務器名IP說明

zk01	192.168.88.21	ZooKeeper節點
zk02	192.168.88.22	ZooKeeper節點
zk03	192.168.88.23	ZooKeeper節點

部署過程參考：https://ken.io/note/zookeeper-cluster-deploy-guide

二、部署過程

1、應用&數據目錄

#創建應用目錄 mkdir /usr/kafka#創建Kafka數據目錄 mkdir /kafka mkdir /kafka/logs chmod 777 -R /kafka

2、下載&解壓

Kafka官方下載地址：https://kafka.apache.org/downloads
這次我下載的是2.3.0版本

#創建并進入下載目錄 mkdir /home/downloads cd /home/downloads#下載安裝包 wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz #解壓到應用目錄 tar -zvxf kafka_2.12-2.3.0.tgz -C /usr/kafka

kafka_2.12-2.3.0.tgz 其中2.12是Scala編譯器的版本，2.3.0才是Kafka的版本

3、Kafka節點配置

#進入應用目錄 cd /usr/kafka/kafka_2.12-2.3.0/#修改配置文件 vi config/server.properties

通用配置

配置日志目錄、指定ZooKeeper服務器

# A comma separated list of directories under which to store log files log.dirs=/kafka/logs# root directory for all kafka znodes. zookeeper.connect=192.168.88.21:2181,192.168.88.22:2181,192.168.88.23:2181

分節點配置

• Kafka01

broker.id=0#listeners=PLAINTEXT://:9092 listeners=PLAINTEXT://192.168.88.51:9092

• Kafka02

broker.id=1#listeners=PLAINTEXT://:9092 listeners=PLAINTEXT://192.168.88.52:9092

• Kafka03

broker.id=2#listeners=PLAINTEXT://:9092 listeners=PLAINTEXT://192.168.88.53:9092

4、防火墻配置

#開放端口 firewall-cmd --add-port=9092/tcp --permanent#重新加載防火墻配置 firewall-cmd --reload

5、啟動Kafka

#進入kafka根目錄 cd /usr/kafka/kafka_2.12-2.3.0/ #啟動 bin/zookeeper-server-start.sh -daemon config/zookeeper.properties bin/kafka-server-start.sh config/server.properties &#啟動成功輸出示例(最后幾行) [2019-06-26 21:48:57,183] INFO Kafka commitId: fc1aaa116b661c8a (org.apache.kafka.common.utils.AppInfoParser) [2019-06-26 21:48:57,183] INFO Kafka startTimeMs: 1561531737175 (org.apache.kafka.common.utils.AppInfoParser) [2019-06-26 21:48:57,185] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

三、Kafka測試

1、創建Topic

在kafka01(Broker)上創建測試Tpoic：test-ken-io，這里我們指定了3個副本、1個分區

bin/kafka-topics.sh --create --bootstrap-server 192.168.88.51:9092 --replication-factor 3 --partitions 1 --topic test-ken-io

Topic在kafka01上創建后也會同步到集群中另外兩個Broker：kafka02、kafka03

2、查看Topic

我們可以通過命令列出指定Broker的

bin/kafka-topics.sh --list --bootstrap-server 192.168.88.52:9092

3、發送消息

這里我們向Broker(id=0)的Topic=test-ken-io發送消息

bin/kafka-console-producer.sh --broker-list 192.168.88.51:9092 --topic test-ken-io#消息內容 > test by ken.io

4、消費消息

在Kafka02上消費Broker03的消息

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

在Kafka03上消費Broker02的消息

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

然后均能收到消息

test by ken.io

這是因為這兩個消費消息的命令是建立了兩個不同的Consumer
如果我們啟動Consumer指定Consumer Group Id就可以作為一個消費組協同工，1個消息同時只會被一個Consumer消費到

bin/kafka-console-consumer.sh --bootstrap-server 192.168.88.53:9092 --topic test-ken-io --from-beginning --group testgroup_kenbin/kafka-console-consumer.sh --bootstrap-server 192.168.88.52:9092 --topic test-ken-io --from-beginning --group testgroup_ken

四、備注

1、Kafka常用配置項說明

Kafka常用Broker配置說明：

配置項默認值/示例值說明

broker.id	0	Broker唯一標識
listeners	PLAINTEXT://192.168.88.53:9092	監聽信息，PLAINTEXT表示明文傳輸
log.dirs	kafka/logs	kafka數據存放地址，可以填寫多個。用”,”間隔
message.max.bytes	message.max.bytes	單個消息長度限制，單位是字節
num.partitions	1	默認分區數
log.flush.interval.messages	Long.MaxValue	在數據被寫入到硬盤和消費者可用前最大累積的消息的數量
log.flush.interval.ms	Long.MaxValue	在數據被寫入到硬盤前的最大時間
log.flush.scheduler.interval.ms	Long.MaxValue	檢查數據是否要寫入到硬盤的時間間隔。
log.retention.hours	24	控制一個log保留時間，單位：小時
zookeeper.connect	192.168.88.21:2181	ZooKeeper服務器地址，多臺用”,”間隔

2、附錄

• https://kafka.apache.org/
• https://zh.wikipedia.org/zh-cn/Kafka

總結

以上是生活随笔為你收集整理的Kafka集群部署CentOS 7的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。