Kafka學習之（六）搭建kafka集羣

想要搭建kafka集羣，必須具有zookeeper集羣，關於zookeeper集羣的搭建，在Kafka學習之（五）搭建kafka集羣之Zookeeper集羣搭建博客有說明。須要具有兩臺以上裝有zookeeper的服務器，其次，穩定可靠的kafka版本，本篇使用kafka_2.11-0.9.0.1。

準備

192.168.244.13一、192.168.244.13三、192.168.244.134，kafka下載地址 https://archive.apache.org/dist/kafka/0.9.0.1/kafka_2.11-0.9.0.1.tgz（wget下載）

建立目錄並解壓

我習慣把壓縮包下載到/usr/local/src下，而後我把下載好的kafka tag包作了目錄拷貝。

# cd /usr/local # mkdir kafka # cd kafka # mkdir kafkalogs # cp /usr/local/kafka......tar.gz /usr/local/kafka

編輯配置文件

cd /usr/local/kafka/kafka_2.11-0.9.0.1/config/ //進入config目錄

主要關注：server.properties 這個文件便可，咱們能夠發如今目錄下：
有不少文件，這裏能夠發現有Zookeeper文件，咱們能夠根據Kafka內帶的zk集羣來啓動，可是建議使用獨立的zk集羣。

broker.id=0 #當前機器在集羣中的惟一標識，和zookeeper的myid性質同樣 port=19092 #當前kafka對外提供服務的端口默認是9092，這裏改大 host.name=192.168.244.131 #這個參數默認是關閉的，在0.8.1有個bug，DNS解析問題，在解析的時候會解析成ip，這裏會有失敗率。 num.network.threads=3 #這個是borker進行網絡處理的線程數 num.io.threads=8 #這個是borker進行I/O處理的線程數 log.dirs=/usr/local/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.244.131:12181,192.168.244.133:12181,192.168.244.134:12181 #設置zookeeper的鏈接端口

上面的參數只是做爲了解一下，實際搭建中須要根據場景來配置。

如今只須要對每臺機器的配置文件作以下更改

#broker.id=0 //每臺服務器的broker.id都不能相同，從0開始，分別0、一、2 #listeners=PLAINTEXT://:9092  #hostname host.name=192.168.244.131 //分別對應ip  log.dirs=/usr/local/kafka/kafkalogs //日誌目錄  #在log.retention.hours=168 //下面新增下面三項 message.max.byte=5242880 default.replication.factor=2 replica.fetch.max.bytes=5242880 #設置zookeeper的鏈接端口 zookeeper.connect=192.168.244.131:12181,192.168.244.133:12181,192.168.244.134:12181

啓動服務

在啓動kafka以前，確保每臺機器的防火牆關閉，其次zookeeper集羣也是啓動中，確保以上條件成立而後執行命令啓動kafka

#./kafka-server-start.sh -daemon ../config/server.properties //在kafka的目錄下bin目錄執行。 表示後臺以server.properties配置文件執行。 三臺分別執行

# jps //檢查服務是否啓動 3506 Jps 3443 Kafka 3397 QuorumPeerMain

建立Topic來驗證是否建立成功

這是官方文檔說明：http://kafka.apache.org/quickstart

//在192.168.244.131 建立Topic # ./kafka-topics.sh --create --zookeeper 192.168.244.131:12181 --replication-factor 2 --partitions 1 --topic wt645631686 Created topic "wt645631686". //解釋 --replication-factor 2 #複製兩份 --partitions 1 #建立1個分區 --topic #主題爲wt645631686

//分別在192.168.244.133和192.168.244.134上建立訂閱者 # ./kafka-console-consumer.sh --zookeeper 192.168.244.134:12181 --topic wt645631686

//在192.168.244.131上建立發佈者 # ./kafka-console-producer.sh --broker-list 192.168.244.131:19092 --topict 645631686

在192.168.244.131上發佈測試，分別觀察另外兩臺機器。

topic其餘說明

# ./kafka-topics.sh --list --zookeeper localhost:12181 //顯示咱們建立的全部topic

# ./kafka-topics.sh --describe --zookeeper 192.168.244.131:12181 --topic wt645631686 //查看指定topic狀況 Topic:wt645631686 PartitionCount:1 //一個分區 ReplicationFactor:2 Configs: Topic: wt645631686 Partition: 0 Leader: 2 Replicas: 2,0 Isr: 2,0

到這裏kafka集羣就搭建完了~

日誌說明

默認kafka的日誌是保存在/opt/kafka/kafka_2.10-0.9.0.0/logs目錄下的，這裏說幾個須要注意的日誌

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

查看zookeeper狀況

進入zookeeper的項目目錄

# pwd /usr/local/zookeeper/zookeeper-3.4.6/bin # ./zkCli.sh -server 127.0.0.1:12181 //默認是不用加’-server‘參數的由於咱們修改了他的端口

Connecting to 127.0.0.1:12181 2018-02-15 15:12:57,356 [myid:] - INFO [main:Environment@100] - Client environment:zookeeper.version=3.4.6-1569965, built on 02/20/2014 09:09 GMT 2018-02-15 15:12:57,360 [myid:] - INFO [main:Environment@100] - Client environment:host.name=localhost 2018-02-15 15:12:57,360 [myid:] - INFO [main:Environment@100] - Client environment:java.version=1.8.0_91 2018-02-15 15:12:57,363 [myid:] - INFO [main:Environment@100] - Client environment:java.vendor=Oracle Corporation 2018-02-15 15:12:57,363 [myid:] - INFO [main:Environment@100] - Client environment:java.home=/usr/local/jdk1.8.0_91/jre 2018-02-15 15:12:57,363 [myid:] - INFO [main:Environment@100] - Client environment:java.class.path=/usr/local/zookeeper/zookeeper-3.4.6/bin/../build/classes:/usr/local/zookeeper/zookeeper-3.4.6/bin/../build/lib/*.jar:/usr/local/zookeeper/zookeeper-3.4.6/bin/../lib/slf4j-log4j12-1.6.1.jar:/usr/local/zookeeper/zookeeper-3.4.6/bin/../lib/slf4j-api-1.6.1.jar:/usr/local/zookeeper/zookeeper-3.4.6/bin/../lib/netty-3.7.0.Final.jar:/usr/local/zookeeper/zookeeper-3.4.6/bin/../lib/log4j-1.2.16.jar:/usr/local/zookeeper/zookeeper-3.4.6/bin/../lib/jline-0.9.94.jar:/usr/local/zookeeper/zookeeper-3.4.6/bin/../zookeeper-3.4.6.jar:/usr/local/zookeeper/zookeeper-3.4.6/bin/../src/java/lib/*.jar:/usr/local/zookeeper/zookeeper-3.4.6/bin/../conf:.:/usr/local/jdk/lib/dt.jar:/usr/local/jdk/lib/tools.jar 2018-02-15 15:12:57,363 [myid:] - INFO [main:Environment@100] - Client environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib 2018-02-15 15:12:57,363 [myid:] - INFO [main:Environment@100] - Client environment:java.io.tmpdir=/tmp 2018-02-15 15:12:57,364 [myid:] - INFO [main:Environment@100] - Client environment:java.compiler=<NA> 2018-02-15 15:12:57,364 [myid:] - INFO [main:Environment@100] - Client environment:os.name=Linux 2018-02-15 15:12:57,364 [myid:] - INFO [main:Environment@100] - Client environment:os.arch=amd64 2018-02-15 15:12:57,365 [myid:] - INFO [main:Environment@100] - Client environment:os.version=2.6.32-358.el6.x86_64 2018-02-15 15:12:57,365 [myid:] - INFO [main:Environment@100] - Client environment:user.name=root 2018-02-15 15:12:57,366 [myid:] - INFO [main:Environment@100] - Client environment:user.home=/root 2018-02-15 15:12:57,366 [myid:] - INFO [main:Environment@100] - Client environment:user.dir=/usr/local/zookeeper/zookeeper-3.4.6/bin 2018-02-15 15:12:57,368 [myid:] - INFO [main:ZooKeeper@438] - Initiating client connection, connectString=127.0.0.1:12181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@799f7e29 Welcome to ZooKeeper! 2018-02-15 15:12:57,404 [myid:] - INFO [main-SendThread(127.0.0.1:12181):ClientCnxn$SendThread@975] - Opening socket connection to server 127.0.0.1/127.0.0.1:12181. Will not attempt to authenticate using SASL (unknown error) JLine support is enabled 2018-02-15 15:12:57,481 [myid:] - INFO [main-SendThread(127.0.0.1:12181):ClientCnxn$SendThread@852] - Socket connection established to 127.0.0.1/127.0.0.1:12181, initiating session 2018-02-15 15:12:57,496 [myid:] - INFO [main-SendThread(127.0.0.1:12181):ClientCnxn$SendThread@1235] - Session establishment complete on server 127.0.0.1/127.0.0.1:12181, sessionid = 0x167509a9cdd0004, negotiated timeout = 30000 WATCHER:: WatchedEvent state:SyncConnected type:None path:null
[zk: 127.0.0.1:12181(CONNECTED) 0] 

[zk: 127.0.0.1:12181(CONNECTED) 1] ls / //查看目錄狀況 執行「ls /」 [controller, controller_epoch, brokers, zookeeper, admin, isr_change_notification, consumers, config] //上面的顯示結果中：只有zookeeper是，zookeeper原生的，其餘都是Kafka建立的

[zk: 127.0.0.1:12181(CONNECTED) 12] ls / [controller, controller_epoch, brokers, zookeeper, admin, isr_change_notification, consumers, config] [zk: 127.0.0.1:12181(CONNECTED) 13] ls /brokers/topics [wt645631686] [zk: 127.0.0.1:12181(CONNECTED) 14] ls /brokers/topics/wt645631686 [partitions] [zk: 127.0.0.1:12181(CONNECTED) 15] ls /brokers/topics/wt645631686/partitions [0] [zk: 127.0.0.1:12181(CONNECTED) 16] get /brokers/ids/0 {"jmx_port":-1,"timestamp":"1543245597011","endpoints":["PLAINTEXT://192.168.244.131:19092"],"host":"192.168.244.131","version":2,"port":19092} cZxid = 0x200000017 ctime = Mon Nov 26 07:19:57 PST 2018 mZxid = 0x200000017 mtime = Mon Nov 26 07:19:57 PST 2018 pZxid = 0x200000017 cversion = 0 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x167509a9cdd0000 dataLength = 143 numChildren = 0

[zk: 127.0.0.1:12181(CONNECTED) 17] get /brokers/topics/wt645631686/partitions/0 //查看partion null cZxid = 0x200000029 ctime = Mon Nov 26 07:29:16 PST 2018 mZxid = 0x200000029 mtime = Mon Nov 26 07:29:16 PST 2018 pZxid = 0x20000002a cversion = 1 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 0 numChildren = 1

暫時完畢~