Kafka——集群安裝部署(自帶Zookeeper)

 

 

kafka簡介

kafka官網：http://kafka.apache.org/
kafka下載頁面：http://kafka.apache.org/downloads
kafka配置快速入門：http://kafka.apache.org/quickstart
新版本的kafka自帶有zookeeper，本篇文章記錄使用自帶zookeeper搭建kafka集群。

為了實現高可用，高吞吐量，kafka以集群方式部署，分別在3臺服務器上安裝部署，因kafka依賴於zookeeper，如果想要使用Kafka，就必須安裝ZK，Kafka中的消費偏置信息、kafka集群、topic信息會被存儲在ZK中。

這裏採用kafka內置的zookeeper，也是以集群方式部署，無需在單獨安裝zookeeper。

 

首先讓我們看幾個基本的消息系統術語：

•Kafka將消息以topic為單位進行歸納。
•將向Kafka topic發布消息的程序成為producers.
•將預訂topics並消費消息的程序成為consumer.
•Kafka以集群的方式運行，可以由一個或多個服務組成，每個服務叫做一個broker.
producers通過網絡將消息發送到Kafka集群，集群向消費者提供消息，如下圖所示：

客戶端和服務端通過TCP協議通信。Kafka提供了Java客戶端，並且對多種語言都提供了支持。

 

環境准備

kafka版本：kafka_2.11-1.0.0.tgz
三台主機IP：172.16.10.91、172.16.10.92、172.16.10.93

 

安裝配置工作

1、kafka安裝

　　下載后解壓kafka，將其解壓到/usr/local目錄下，刪除壓縮包即可：

tar xzvf kafka_2.11-1.0.0.tgz  #解壓

　　進入到kafka的config目錄 

 

　　自定義目錄：

首先新建kafka的日志目錄和zookeeper數據目錄，因為這兩項默認放在tmp目錄，而tmp目錄中內容會隨重啟而丟失,所以我們自定義以下目錄:

mkdir /usr/local/kafka_2.11-1.0.0/zookeeper        #創建zookeeper數據目錄
mkdir /usr/local/kafka_2.11-1.0.0/log              #創建日志目錄
mkdir /usr/local/kafka_2.11-1.0.0/log/zookeeper    #創建zookeeper日志目錄
mkdir /usr/local/kafka_2.11-1.0.0/log/kafka        #創建kafka日志目錄

 

2、zookeeper配置

　　> 修改 zookeeper.properties

進入config目錄下，修改關鍵配置如下，3台服務器的zookeeper.properties配置文件都一樣

#修改為自定義的zookeeper數據目錄
dataDir=/usr/local/kafka_2.11-1.0.0/zookeeper

#修改為自定義的zookeeper日志目錄
dataLogDir=/usr/local/kafka_2.11-1.0.0/log/zookeeper 

# 端口
clientPort=2181

#注釋掉
#maxClientCnxns=0

#設置連接參數，添加如下配置
tickTime=2000　　　　#為zk的基本時間單元，毫秒
initLimit=10　　　　 #Leader-Follower初始通信時限 tickTime*10
syncLimit=5　　　　　#Leader-Follower同步通信時限 tickTime*5

#設置broker Id的服務地址
server.0=172.16.10.91:2888:3888
server.1=172.16.10.92:2888:3888
server.2=172.16.10.93:2888:3888

 

　　> zookeeper數據目錄添加myid配置

在各台服務器的zookeeper數據目錄【/usr/local/kafka_2.11-1.0.0/zookeeper】添加myid文件，寫入服務broker.id屬性值                                                                   

如這里的目錄是/usr/local/kafka_2.11-1.0.0/zookeeper

注：3臺服務器都要添加myid文件，但值的內容不一樣，具體關係請見下圖：

 

例如：第一台broker.id為0的服務到該目錄【/usr/local/kafka_2.11-1.0.0/zookeeper】下執行以下命令，會生成myid文件內容

[root@mmc config]# echo 0 > myid

 

 

3、kafka配置

 　　進入config目錄下，修改server.properties文件

注：3臺服務器都要配置，其中【broker.id、advertised.listeners】不一樣，其它配置都一樣

############################# Server Basics #############################
# broker 的全局唯一編號，不能重復
broker.id=1


############################# Socket Server Settings #############################
# 配置監聽,修改為本機ip
advertised.listeners=PLAINTEXT://172.16.10.91:9092   

# 處理網絡請求的線程數量，默認
num.network.threads=3

# 用來處理磁盤IO的線程數量，默認
num.io.threads=8

# 發送套接字的緩沖區大小，默認
socket.send.buffer.bytes=102400

# 接收套接字的緩沖區大小，默認
socket.receive.buffer.bytes=102400

# 請求套接字的緩沖區大小，默認
socket.request.max.bytes=104857600


############################# Zookeeper #############################
# 配置三台服務zookeeper連接地址
zookeeper.connect=172.16.10.91:2181,172.16.10.92:2181,172.16.10.93:2181


############################# Log Basics #############################
# kafka 運行日志存放路徑
log.dirs=/usr/local/kafka_2.11-1.0.0/log/kafka 
# topic 在當前broker上的分片個數，與broker保持一致
num.partitions=3

# 用來恢復和清理data下數據的線程數量，默認
num.recovery.threads.per.data.dir=1


############################# Log Retention Policy #############################
# segment文件保留的最長時間，超時將被刪除，默認
log.retention.hours=168

# 滾動生成新的segment文件的最大時間，默認
log.roll.hours=168

 

 4、kafka啟動

　　kafka啟動時先啟動zookeeper，再啟動kafka；關閉時相反，先關閉kafka，再關閉zookeeper

• 進入kafka根目錄

[root@localhost kafka_2.11-1.0.0]# cd /usr/local/kafka_2.11-1.0.0/

 

• 啟動zookeeper

　　輸入命令：

[root@localhostkafka_2.11-1.0.0]# bin/zookeeper-server-start.sh config/zookeeper.properties &

#後臺啟動命令：
[root@localhostkafka_2.11-1.0.0]# nohup bin/zookeeper-server-start.sh config/zookeeper.properties >log/zookeeper/zookeeper.log 2>1 &

 

　　查看log/zookeeper/zookeeper.log 文件，第1、2臺服務器啟動的時候，會報錯Connection refused，因為另外1、2臺服務器還沒啟動，先不用管它

 

　　查看log/zookeeper/zookeeper.log 文件，等3臺服務都啟動完後，就不在報錯了

 

• 啟動kafka

　　輸入命令：

[root@localhostkafka_2.11-1.0.0]# bin/kafka-server-start.sh config/server.properties &

#後臺啟動命令：
[root@localhostkafka_2.11-1.0.0]# nohup bin/kafka-server-start.sh config/server.properties >log/kafka/kafka.log 2>1 &

　　查看log/kafka/kafka.log 文件，啟動過程中沒有報錯，且有以下信息輸出，說明啟動成功

 

 

 5、kafka測試驗證

　　分別啟動三台機器的zookeeper，三台機器的zookeeper都啟動成功后，再分別啟動三台機器的kafka。

• 進入kafka根目錄

[root@localhost kafka_2.11-1.0.0]# cd /usr/local/kafka_2.11-1.0.0/

• 在某臺機器創建topic，名稱為test

[root@localhost kafka_2.11-1.0.0]# bin/kafka-topics.sh -create --zookeeper 172.16.10.91:2181,172.16.10.92:2181,172.16.10.93:2181 -replication-factor 3 --partitions 3 --topic test

命令解析： 
--create：　　　　　　　　　 #指定創建topic動作 

--topic：　　　　　　　　　　#指定新建topic的名稱 

--zookeeper： 　　　　　　　#指定kafka連接zk的連接url，該值和server.properties文件中的配置項{zookeeper.connect}一樣

--partitions 3　　　　　　 #指定當前創建的kafka分區數量，默認為1個

--replication-factor 3　　#指定每個分區的副本集數量，不能大於 broker 節點數量，多了也沒用，1個節點放>=2個副本掛了都完蛋。

 

• 查看topic，確認topic創建成功

[root@localhost kafka_2.11-1.0.0]# bin/kafka-topics.sh --list --zookeeper 172.16.10.91:2181

 

• 查看topic，詳細信息

[root@localhost kafka_2.11-1.0.0]# bin/kafka-topics.sh --describe --zookeeper 172.16.10.93:2181 --topic test

可以描述為：Topic分區數/副本數/副本Leader/副本ISR等信息： 

　　“leader”：該節點負責該分區的所有的讀和寫，每個節點的leader都是隨機選擇的。
　　“replicas”：備份的節點列表，無論該節點是否是leader或者目前是否還活着，只是顯示。
　　“isr”：同步備份”的節點列表，也就是活着的節點並且正在同步leader
  　　　其中Replicas和Isr中的1,2,0就對應着3個broker他們的broker.id屬性！

 

• 在某臺服務器上創建生產者

[root@localhost kafka_2.11-1.0.0]# bin/kafka-console-producer.sh --broker-list 172.16.10.91:9092,172.16.10.92:9092,172.16.10.93:9092 --topic test

 

• 在另外兩台服務器創建消費者

[root@localhost kafka_2.11-1.0.0]# bin/kafka-console-consumer.sh --zookeeper 172.16.10.91:2181,172.16.10.92:2181,172.16.10.93:2181 --topic test --from-beginning

　　

• 生產者服務器截圖

 

• 消費者服務器截圖

 

• 修改topic信息

bin/kafka-topics.sh --zookeeper 192.168.187.146:2181 --alter --topic test --config max.message.bytes=128000
bin/kafka-topics.sh --zookeeper 192.168.187.146:2181 --alter --topic test --delete-config max.message.bytes
bin/kafka-topics.sh --zookeeper 192.168.187.146:2181 --alter --topic test --partitions 10 
bin/kafka-topics.sh --zookeeper 192.168.187.146:2181 --alter --topic test --partitions 3 ## Kafka分區數量只允許增加，不允許減少

 

• 刪除topic

注意，只是刪除Topic在zk的元數據，日志數據仍需手動刪除

[root@localhost kafka_2.11-1.0.0]# bin/kafka-topics.sh -delete --zookeeper 172.16.10.91:2181,172.16.10.92:2181,172.16.10.93:2181 --topic test

Note: This will have no impact if delete.topic.enable is not set to true.## 默認情況下，刪除是標記刪除，沒有實際刪除這個Topic；如果運行刪除Topic，兩種方式：
  方式一：通過delete命令刪除后，手動將本地磁盤以及zk上的相關topic的信息刪除即可
  方式二：配置server.properties文件，給定參數delete.topic.enable=true，重啟kafka服務，此時執行delete命令表示允許進行Topic的刪除

 

• 查看kafka未消費的消息

[root@localhost kafka_2.11-1.0.0]# bin/kafka-console-consumer.sh --bootstrap-server 172.16.10.91:9092,172.16.10.92:9092,172.16.10.93:9092 --topic logstash-busi-2 --from-beginning

 

• 查看kafka消費進度

命令：bin/kafka-consumer-groups.sh --bootstrap-server <Kafka broker 連接信息> --describe --group <group 名稱>

bin/kafka-consumer-groups.sh --bootstrap-server 172.16.10.91:9092,172.16.10.92:9092,172.16.10.93:9092 --describe --group terminalMonitor

 

 

 

 

 6、kafka測試集群的容錯能力
　　Kafka是支持容錯的，下面我們來對Kafka的容錯性進行測試，測試步驟如下

(1).查看topic的詳細信息，觀察那個blocker的角色是leader，那些blocker的角色是follower

[hadoop@Master ~]$ kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:
    Topic: my-replicated-topic  Partition: 0    Leader: 2   Replicas: 2,3,1 Isr: 2,3,1

(2).手工kill掉任意一個狀態是follower的borker，測試生成和消費信息是否正確

步驟1中可以看到 2 為leader,1 和 3為 follower，將follower為1的進程kill掉

 

啟動生產和消費者測試信息是否正確

 

 

結論：kill掉任意一個狀態是follower的broker，生成和消費信息正確，不受任何影響

 

(3).手工kill掉狀態是leader的borker，測試生產和消費的信息是否正確

borker2的角色為leader，將它kill掉，borker 3變成了leader

啟動生產和消費者測試信息是否正確

 

 

結論：kill掉狀態是leader的borker，生產和消費的信息正確

總結：不管當前狀態的borker是leader還是follower，當我們kill掉后，只要有一個borker能夠正常使用，則消息仍然能夠正常的生產和發送。即Kafka的容錯性是有保證的！