kafka的安裝以及基本用法

kafka的安裝

kafka依賴於ZooKeeper,因此在運行kafka以前須要先部署ZooKeeper集羣，ZooKeeper集羣部署方式分爲兩種,一種是單獨部署(推薦),另一種是使用kafka自帶的。
這裏咱們默認已經部署好了ZooKeeper集羣

• 下載kafka(3臺服務器)

cd /usr/local/src wget http://apache.opencas.org/kafka/2.1.0/kafka_2.11-2.1.0.tgz tar -zxvf kafka_2.11-2.1.0.tgz mv /usr/local/src/kafka_2.11-2.1.0 /usr/local/kafka cd /usr/local/src wget http://apache.opencas.org/kafka/2.1.0/kafka_2.11-2.1.0.tgz tar -zxvf kafka_2.11-2.1.0.tgz mv /usr/local/src/kafka_2.11-2.1.0 /usr/local/kafka

• 修改配置文件

主要關注：server.properties 這個文件便可
cd /usr/local/kafka

配置文件詳情解釋

broker.id=0  #當前機器在集羣中的惟一標識，和zookeeper的myid性質同樣
port=19092 #當前kafka對外提供服務的端口默認是9092
host.name=192.168.7.100 #這個參數默認是關閉的，在0.8.1有個bug，DNS解析問題，失敗率的問題。
num.network.threads=3 #這個是borker進行網絡處理的線程數
num.io.threads=8 #這個是borker進行I/O處理的線程數
log.dirs=/opt/kafka/kafkalogs/ #消息存放的目錄，這個目錄能夠配置爲「，」逗號分割的表達式，上面的num.io.threads要大於這個目錄的個數這個目錄，若是配置多個目錄，新建立的topic他把消息持久化的地方是，當前以逗號分割的目錄中，那個分區數最少就放那一個
socket.send.buffer.bytes=102400 #發送緩衝區buffer大小，數據不是一會兒就發送的，先回存儲到緩衝區了到達必定的大小後在發送，能提升性能
socket.receive.buffer.bytes=102400 #kafka接收緩衝區大小，當數據到達必定大小後在序列化到磁盤
socket.request.max.bytes=104857600 #這個參數是向kafka請求消息或者向kafka發送消息的請請求的最大數，這個值不能超過java的堆棧大小
num.partitions=1 #默認的分區數，一個topic默認1個分區數
log.retention.hours=168 #默認消息的最大持久化時間，168小時，7天
message.max.byte=5242880  #消息保存的最大值5M
default.replication.factor=2  #kafka保存消息的副本數，若是一個副本失效了，另外一個還能夠繼續提供服務
replica.fetch.max.bytes=5242880  #取消息的最大直接數
log.segment.bytes=1073741824 #這個參數是：由於kafka的消息是以追加的形式落地到文件，當超過這個值的時候，kafka會新起一個文件
log.retention.check.interval.ms=300000 #每隔300000毫秒去檢查上面配置的log失效時間（log.retention.hours=168 ），到目錄查看是否有過時的消息若是有，刪除
log.cleaner.enable=false #是否啓用log壓縮，通常不用啓用，啓用的話能夠提升性能
zookeeper.connect=192.168.7.100:12181,192.168.7.101:12181,192.168.7.107:1218 #設置zookeeper的鏈接端口

實際配置參數以下(具體根據實際狀況作修改)

broker.id=0
#listeners=PLAINTEXT://:9092 -----》註釋掉修改成以下
port=9092
host.name=192.168.94.130  (每臺服務器地址須要單獨修改)
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.94.130:12181,192.168.94.131:12181,192.168.94.132:12181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

• 啓動(3臺服務器上執行)

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

• 查看

[root@localhost ~]# jps
1745 QuorumPeerMain
3564 Jps
3501 Kafka

這樣就安裝完成了

kafka的基本使用

• 建立一個名爲「test」的Topic，只有一個分區和一個備份

bin/kafka-topics.sh --create --zookeeper 192.168.94.130:12181，192.168.94.131:12181，192.168.94.132:12181 --replication-factor 1 --partitions 1 --topic test

• 查看已建立的topic信息：

./kafka-topics.sh --list --zookeeper 192.168.94.130:12181,192.168.94.131:12181,192.168.94.132:12181

測試發送消息

`

生產者
./kafka-console-producer.sh --broker-list 192.168.94.130:9092,192.168.94.131:9092,192.168.94.132:9092 --topic shuaige

消費者
./kafka-console-consumer.sh --bootstrap-server 192.168.94.132:9092,192.168.94.131:9092,192.168.94.130:9092 --topic shuaige  --from-beginning

查看不一樣topic下的broker信息

./kafka-topics.sh --describe --zookeeper 192.168.94.130:12181,192.168.94.131:12181,192.168.94.132:12181 --topic first

[root@localhost bin]# ./kafka-topics.sh --describe --zookeeper 192.168.94.130:12181,192.168.94.131:12181,192.168.94.132:12181--topic first
Topic:first     PartitionCount:1        ReplicationFactor:3     Configs:
        Topic: first    Partition: 0    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3
[root@localhost bin]# ./kafka-topics.sh --describe --zookeeper 192.168.94.130:12181,192.168.94.131:12181,192.168.94.132:12181 --topic second
Topic:second    PartitionCount:2        ReplicationFactor:3     Configs:
        Topic: second   Partition: 0    Leader: 3       Replicas: 3,1,2 Isr: 3,2,1
        Topic: second   Partition: 1    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3

相關信息解釋

是輸出解釋。第一行給出了各個分區的概況，分區有幾個就有幾行分區詳細信息介紹。（我建立了兩個topic，一個是first，只有一個分區；一個是second，兩個分區）

• Leader：是負責當前分區的全部讀寫請求。每一個節點都將領導一個隨機選擇的分區。
• Replicas ：是節點列表，複製分區日誌，無論他們是否是Leader或者無論它們是否還活着。
• Isr：是in-sync的集合。這是Replicas列表當前還活着的子集。

相關日誌說明

server.log #kafka的運行日誌
state-change.log #kafka他是用zookeeper來保存狀態，因此他可能會進行切換，切換的日誌就保存在這裏
controller.log #kafka選擇一個節點做爲「controller」,當發現有節點down掉的時候它負責在游泳分區的全部節點中選擇新的leader,這使得Kafka能夠批量的高效的管理全部分區節點的主從關係。若是controller down掉了，活着的節點中的一個會備切換爲新的controller.

常見錯誤：
在消費或者在生產者地方報錯
`

[root@localhost bin]# ./kafka-console-consumer.sh --bootstrap-server 192.168.94.130:9092 --topic shuaige --from-beginning
[2019-04-06 23:36:40,431] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 10 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:40,557] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 11 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:40,682] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 12 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:40,805] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 13 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:40,936] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 14 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:41,064] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 15 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:41,198] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 16 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:41,424] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 19 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:42,109] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 31 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:42,236] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 32 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:42,356] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 33 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:42,727] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 41 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:42,954] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 44 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:43,080] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 45 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:43,214] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 46 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:43,340] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 47 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:43,455] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 48 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:43,984] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 60 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:44,108] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 61 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:44,233] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with correlation id 62 : {shuaige=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2019-04-06 23:36:44,574] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20606] Error while fetching metadata with c

解決辦法

通過反覆測試驗證環境配置信息，最終參考了他人的經驗，是kafka的server.properties的配置錯誤。主要是下面的內容配置有問題：listeners=PLAINTEXT://:9092
將這句註釋掉，而後在配置文件中添加下面的兩行配置，
指明當前broker的地址：port=9092
host.name=192.168.94.130 #依據具體的服務器，配置相應的服務器的IP地址便可。
(3臺服務器都須要修改)