Kafka初識
1、Kafka使用背景
在我們大量使用分布式數據庫、分布式計算集群的時候，是否會遇到這樣的一些問題：
- 我們想分析下用戶行為（pageviews），以便我們設計出更好的廣告位
- 我想對用戶的搜索關鍵詞進行統計，分析出當前的流行趨勢
- 有些數據，存儲數據庫浪費，直接存儲硬盤效率又低
這些場景都有一個共同點：
數據是由上游模塊產生，上游模塊，使用上游模塊的數據計算、統計、分析，這個時候就可以使用消息系統，尤其是分布式消息系統！
2、Kafka的定義
What is Kafka：它是一個分布式消息系統，由linkedin使用scala編寫，用作LinkedIn的活動流（Activity Stream）和運營數據處理管道（Pipeline）的基礎。具有高水平擴展和高吞吐量。
3、Kafka和其他主流分布式消息系統的對比
定義解釋：
1、Java 和 scala都是運行在JVM上的語言。
2、erlang和最近比較火的和go語言一樣是從代碼級別就支持高並發的一種語言，所以RabbitMQ天生就有很高的並發性能，但是 有RabbitMQ嚴格按照AMQP進行實現，受到了很多限制。kafka的設計目標是高吞吐量，所以kafka自己設計了一套高性能但是不通用的協議，他也是仿照AMQP（ Advanced Message Queuing Protocol 高級消息隊列協議）設計的。
3、事物的概念：在數據庫中，多個操作一起提交，要么操作全部成功，要么全部失敗。舉個例子， 在轉賬的時候付款和收款，就是一個事物的例子，你給一個人轉賬，你轉成功，並且對方正常行收到款項后，這個操作才算成功，有一方失敗，那么這個操作就是失敗的。
對應消在息隊列中，就是多條消息一起發送，要么全部成功，要么全部失敗。3個中只有ActiveMQ支持，這個是因為，RabbitMQ和Kafka為了更高的性能，而放棄了對事物的支持 。
4、集群：多台服務器組成的整體叫做集群，這個整體對生產者和消費者來說，是透明的。其實對消費系統組成的集群添加一台服務器減少一台服務器對生產者和消費者都是無感之的。
5、負載均衡，對消息系統來說負載均衡是大量的生產者和消費者向消息系統發出請求消息，系統必須均衡這些請求使得每一台服務器的請求達到平衡，而不是大量的請求，落到某一台或幾台，使得這幾台服務器高負荷或超負荷工作，嚴重情況下會停止服務或宕機。
6、動態擴容是很多公司要求的技術之一，不支持動態擴容就意味着停止服務，這對很多公司來說是不可以接受的。
注：
阿里巴巴的Metal,RocketMQ都有Kafka的影子，他們要么改造了Kafka或者借鑒了Kafka，最后Kafka的動態擴容是通過Zookeeper來實現的。
Zookeeper是一種在分布式系統中被廣泛用來作為：分布式狀態管理、分布式協調管理、分布式配置管理、和分布式鎖服務的集群。kafka增加和減少服務器都會在Zookeeper節點上觸發相應的事件kafka系統會捕獲這些事件，進行新一輪的負載均衡，客戶端也會捕獲這些事件來進行新一輪的處理。
Kafka相關概念
1、 AMQP協議
Advanced Message Queuing Protocol （高級消息隊列協議）
The Advanced Message Queuing Protocol (AMQP)：
是一個標准開放的應用層的消息中間件（Message Oriented Middleware）協議。AMQP定義了通過網絡發送的字節流的數據格式。因此兼容性非常好，任何實現AMQP協議的程序都可以和與AMQP協議兼容的其他程序交互，可以很容易做到跨語言，跨平台。
上面說的3種比較流行的消息隊列協議，要么支持AMQP協議，要么借鑒了AMQP協議的思想進行了開發、實現、設計。
2、
一些基本的概念
1、消費者：（Consumer）：從消息隊列中請求消息的客戶端應用程序
2、生產者：（Producer） ：向broker發布消息的應用程序
3、AMQP服務端（broker）：用來接收生產者發送的消息並將這些消息路由給服務器中的隊列，便於fafka將生產者發送的消息，動態的添加到磁盤並給每一條消息一個偏移量，所以對於kafka一個broker就是一個應用程序的實例
kafka支持的客戶端語言：Kafka客戶端支持當前大部分主流語言，包括：C、C++、Erlang、Java、.net、perl、PHP、Python、Ruby、Go、Javascript
可以使用以上任何一種語言和kafka服務器進行通信（即辨析自己的consumer從kafka集群訂閱消息也可以自己寫producer程序）
3、Kafka架構
生產者生產消息、kafka集群、消費者獲取消息這樣一種架構，如下圖：
kafka集群中的消息，是通過Topic（主題）來進行組織的，如下圖：
一些基本的概念：
1、主題（Topic）：一個主題類似新聞中的體育、娛樂、教育等分類概念，在實際工程中通常一個業務一個主題。
2、分區（Partition）：一個Topic中的消息數據按照多個分區組織，分區是kafka消息隊列組織的最小單位，一個分區可以看作是一個FIFO（ First Input First Output的縮寫，先入先出隊列）的隊列。
kafka分區是提高kafka性能的關鍵所在，當你發現你的集群性能不高時，常用手段就是增加Topic的分區，分區里面的消息是按照從新到老的順序進行組織，消費者從隊列頭訂閱消息，生產者從隊列尾添加消息。
備份（Replication）：為了保證分布式可靠性，kafka0.8開始對每個分區的數據進行備份（不同的Broker上），防止其中一個Broker宕機造成分區上的數據不可用。
kafka0.7是一個很大的改變：1、增加了備份2、增加了控制借點概念，增加了集群領導者選舉 。
Zookeeper集群搭建
Kafka集群是把狀態保存在Zookeeper中的，首先要搭建Zookeeper集群。
1、軟件環境
（3台服務器-我的測試）
192.168.7.100 server1
192.168.7.101 server2
192.168.7.107 server3
1、Linux服務器一台、三台、五台、（
2*n+1），Zookeeper集群的工作是超過半數才能對外提供服務，3台中超過兩台超過半數，允許1台掛掉 ，是否可以用偶數，其實沒必要。
如果有四台那么掛掉一台還剩下三台服務器，如果在掛掉一個就不行了，這里記住是超過半數。
2、Java jdk1.7 zookeeper是用java寫的所以他的需要JAVA環境，java是運行在java虛擬機上的
3、Zookeeper的穩定版本Zookeeper 3.4.6版本
2、配置&安裝Zookeeper
下面的操作是：3台服務器統一操作
1、安裝Java
yum list java*
yum -y install java-1.7.0-openjdk*
2、下載Zookeeper
首先要注意在生產環境中目錄結構要定義好，防止在項目過多的時候找不到所需的項目
#我的目錄統一放在/opt下面 #首先創建Zookeeper項目目錄 mkdir zookeeper #項目目錄 cd zookeeper mkidr -p {data,logs}#存放快照,事物日志
下載Zookeeper
#下載軟件 cd /opt/zookeeper/ wget http://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz #解壓軟件 tar -zxf zookeeper-3.4.6.tar.gz
3、修改配置文件
進入到解壓好的目錄里面的conf目錄中，查看
#進入conf目錄 /opt/zookeeper/zookeeper-3.4.6/conf #查看 [root@192.168.7.107]$ ll -rw-rw-r--. 1 1000 1000 535 Feb 20 2014 configuration.xsl -rw-rw-r--. 1 1000 1000 2161 Feb 20 2014 log4j.properties -rw-rw-r--. 1 1000 1000 922 Feb 20 2014 zoo_sample.cfg
#zoo_sample.cfg 這個文件是官方給我們的zookeeper的樣板文件，給他復制一份命名為zoo.cfg，zoo.cfg是官方指定的文件命名規則。
3台服務器的配置文件
tickTime=2000 initLimit=5 syncLimit=2 dataDir=/opt/zookeeper/data dataLogDir=/opt/zookeeper/logs clientPort=2181 # The number of snapshots to retain in dataDir autopurge.snapRetainCount=3 # Purge task interval in hours # Set to "0" to disable auto purge feature autopurge.purgeInterval=1 server.1=10.0.0.201:2888:3888 server.2=10.0.0.202:2888:3888 server.3=10.0.0.203:2888:3888 #server.1 這個1是服務器的標識也可以是其他的數字， 表示這個是第幾號服務器，用來標識服務器，這個標識要寫到快照目錄(data)下面myid文件里 #第一個端口是master和slave之間的通信端口，默認是2888，第二個端口是leader選舉的端口，集群剛啟動的時候選舉或者leader掛掉之后進行新的選舉的端口默認是3888
配置文件解釋：
#tickTime： 這個時間是作為 Zookeeper 服務器之間或客戶端與服務器之間維持心跳的時間間隔，也就是每個 tickTime 時間就會發送一個心跳。 #initLimit： 這個配置項是用來配置 Zookeeper 接受客戶端（這里所說的客戶端不是用戶連接 Zookeeper 服務器的客戶端，而是 Zookeeper 服務器集群中連接到 Leader 的 Follower 服務器）初始化連接時最長能忍受多少個心跳時間間隔數。當已經超過 5個心跳的時間（也就是 tickTime）長度后 Zookeeper 服務器還沒有收到客戶端的返回信息，那么表明這個客戶端連接失敗。總的時間長度就是 5*2000=10 秒 #syncLimit： 這個配置項標識 Leader 與Follower 之間發送消息，請求和應答時間長度，最長不能超過多少個 tickTime 的時間長度，總的時間長度就是5*2000=10秒 #dataDir： 快照日志的存儲路徑 #dataLogDir： 事物日志的存儲路徑，如果不配置這個那么事物日志會默認存儲到dataDir制定的目錄，這樣會嚴重影響zk的性能，當zk吞吐量較大的時候，產生的事物日志、快照日志太多 #clientPort： 這個端口就是客戶端連接 Zookeeper 服務器的端口，Zookeeper 會監聽這個端口，接受客戶端的訪問請求。修改他的端口改大點
創建myid文件
#server1 echo "1" > /opt/zookeeper/data/myid #server2 echo "2" > /opt/zookeeper/data/myid #server3 echo "3" > /opt/zookeeper/data/myid
4、重要配置說明
1、myid文件和server.myid 在快照目錄下存放的標識本台服務器的文件，他是整個zk集群用來發現彼此的一個重要標識。
2、zoo.cfg 文件是zookeeper配置文件 在conf目錄里。
3、log4j.properties文件是zk的日志輸出文件 在conf目錄里用java寫的程序基本上有個共同點日志都用log4j，來進行管理。
# Define some default values that can be overridden by system properties zookeeper.root.logger=INFO, CONSOLE #日志級別 zookeeper.console.threshold=INFO #使用下面的console來打印日志 zookeeper.log.dir=. #日志打印到那里，是咱們啟動zookeeper的目錄 （建議設置統一的日志目錄路徑） zookeeper.log.file=zookeeper.log zookeeper.log.threshold=DEBUG zookeeper.tracelog.dir=. zookeeper.tracelog.file=zookeeper_trace.log # # ZooKeeper Logging Configuration # # Format is "<default threshold> (, <appender>)+ # DEFAULT: console appender only log4j.rootLogger=${zookeeper.root.logger} # Example with rolling log file #log4j.rootLogger=DEBUG, CONSOLE, ROLLINGFILE # Example with rolling log file and tracing #log4j.rootLogger=TRACE, CONSOLE, ROLLINGFILE, TRACEFILE # # Log INFO level and above messages to the console # log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender log4j.appender.CONSOLE.Threshold=${zookeeper.console.threshold} log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n # Add ROLLINGFILE to rootLogger to get log file output # Log DEBUG level and above messages to a log file log4j.appender.ROLLINGFILE=org.apache.log4j.RollingFileAppender log4j.appender.ROLLINGFILE.Threshold=${zookeeper.log.threshold} log4j.appender.ROLLINGFILE.File=${zookeeper.log.dir}/${zookeeper.log.file} # Max log file size of 10MB log4j.appender.ROLLINGFILE.MaxFileSize=10MB # uncomment the next line to limit number of backup files #log4j.appender.ROLLINGFILE.MaxBackupIndex=10 log4j.appender.ROLLINGFILE.layout=org.apache.log4j.PatternLayout log4j.appender.ROLLINGFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L] - %m%n # # Add TRACEFILE to rootLogger to get log file output # Log DEBUG level and above messages to a log file log4j.appender.TRACEFILE=org.apache.log4j.FileAppender log4j.appender.TRACEFILE.Threshold=TRACE log4j.appender.TRACEFILE.File=${zookeeper.tracelog.dir}/${zookeeper.tracelog.file} log4j.appender.TRACEFILE.layout=org.apache.log4j.PatternLayout ### Notice we are including log4j's NDC here (%x) log4j.appender.TRACEFILE.layout.ConversionPattern=%d{ISO8601} [myid:%X{myid}] - %-5p [%t:%C{1}@%L][%x] - %m%n
4、zkEnv.sh和zkServer.sh文件
zkServer.sh 主的管理程序文件
zkEnv.sh 是主要配置，zookeeper集群啟動時配置環境變量的文件
5、還有一個需要注意
ZooKeeper server
will not remove old snapshots and log files when using the default configuration (see autopurge below), this is the responsibility of the operator
zookeeper不會主動的清除舊的快照和日志文件，這個是操作者的責任。
但是可以通過命令去定期的清理。
#!/bin/bash #snapshot file dir dataDir=/opt/zookeeper/zkdata/version-2 #tran log dir dataLogDir=/opt/zookeeper/zkdatalog/version-2 #Leave 66 files count=66 count=$[$count+1] ls -t $dataLogDir/log.* | tail -n +$count | xargs rm -f ls -t $dataDir/snapshot.* | tail -n +$count | xargs rm -f #以上這個腳本定義了刪除對應兩個目錄中的文件，保留最新的66個文件，可以將他寫到crontab中，設置為每天凌晨2點執行一次就可以了。 #zk log dir del the zookeeper log #logDir= #ls -t $logDir/zookeeper.log.* | tail -n +$count | xargs rm -f
其他方法：
第二種：使用ZK的工具類PurgeTxnLog，它的實現了一種簡單的歷史文件清理策略，可以在這里看一下他的使用方法 http://zookeeper.apache.org/doc/r3.4.6/zookeeperAdmin.html
第三種：對於上面這個執行，ZK自己已經寫好了腳本，在bin/zkCleanup.sh中，所以直接使用這個腳本也是可以執行清理工作的。
第四種：從3.4.0開始，zookeeper提供了自動清理snapshot和事務日志的功能，通過配置 autopurge.snapRetainCount 和 autopurge.purgeInterval 這兩個參數能夠實現定時清理了。這兩個參數都是在zoo.cfg中配置的：
autopurge.purgeInterval 這個參數指定了清理頻率，單位是小時，需要填寫一個1或更大的整數，默認是0，表示不開啟自己清理功能。
autopurge.snapRetainCount 這個參數和上面的參數搭配使用，這個參數指定了需要保留的文件數目。默認是保留3個。
推薦使用第一種方法，對於運維人員來說，將日志清理工作獨立出來，便於統一管理也更可控。畢竟zk自帶的一些工具並不怎么給力。
5、啟動服務並查看
1、啟動服務
#進入到Zookeeper的bin目錄下#啟動服務（3台都需要操作） /opt/zookeeper/bin/zkServer.sh start
2、檢查服務狀態
# 檢查zookeeper狀態 /opt/zookeeper/bin/zkServer.sh status
通過status就能看到狀態：
/opt/zookeeper/bin/zkServer.sh status JMX enabled by default Using config: /opt/zookeeper/zookeeper-3.4.6/bin/../conf/zoo.cfg #配置文件 Mode: follower #這台服務器角色
zk集群一般只有一個leader，多個follower，主一般是相應客戶端的讀寫請求，而從主同步數據，當主掛掉之后就會從follower里投票選舉一個leader出來。
可以用“jps”查看zk的進程，這個是zk的整個工程的main
#執行命令jps 20348 Jps 4233 QuorumPeerMain
Kafka集群搭建
1、軟件環境
2、創建目錄並下載安裝軟件
#下載軟件 wget -P /usr/local/src http://archive.apache.org/dist/kafka/0.9.0.0/kafka_2.10-0.9.0.0.tgz #解壓軟件 tar -zxf kafka_2.11-0.9.0.1.tgz / -C /opt #創建目錄 mv /opt/kafka_2.11-0.9.0 /opt/kafka#創建項目目錄 mkdir /opt/kafka/logs #創建kafka消息目錄，主要存放kafka消息
3、修改配置文件
進入到config目錄
cd /opt/kafka/config/
主要關注：server.properties 這個文件即可，我們可以發現在目錄下：
有很多文件，這里可以發現有Zookeeper文件，我們可以根據Kafka內帶的zk集群來啟動，但是建議使用獨立的zk集群
-rw-r--r--. 1 root root 906 Feb 12 08:37 connect-console-sink.properties -rw-r--r--. 1 root root 909 Feb 12 08:37 connect-console-source.properties -rw-r--r--. 1 root root 2110 Feb 12 08:37 connect-distributed.properties -rw-r--r--. 1 root root 922 Feb 12 08:38 connect-file-sink.properties -rw-r--r--. 1 root root 920 Feb 12 08:38 connect-file-source.properties -rw-r--r--. 1 root root 1074 Feb 12 08:37 connect-log4j.properties -rw-r--r--. 1 root root 2055 Feb 12 08:37 connect-standalone.properties -rw-r--r--. 1 root root 1199 Feb 12 08:37 consumer.properties -rw-r--r--. 1 root root 4369 Feb 12 08:37 log4j.properties -rw-r--r--. 1 root root 2228 Feb 12 08:38 producer.properties -rw-r--r--. 1 root root 5699 Feb 15 18:10 server.properties -rw-r--r--. 1 root root 3325 Feb 12 08:37 test-log4j.properties -rw-r--r--. 1 root root 1032 Feb 12 08:37 tools-log4j.properties -rw-r--r--. 1 root root 1023 Feb 12 08:37 zookeeper.properties
修改配置文件：
broker.id=0 #當前機器在集群中的唯一標識，和zookeeper的myid性質一樣 listeners=PLAINTEXT://10.0.0.201:9092 port=9092 #當前kafka對外提供服務的端口默認是9092 host.name=10.0.0.201 #這個參數默認是關閉的，在0.8.1有個bug，DNS解析問題，失敗率的問題。 num.network.threads=3 #這個是borker進行網絡處理的線程數 num.io.threads=8 #這個是borker進行I/O處理的線程數 log.dirs=/opt/kafka/logs/ #消息存放的目錄，這個目錄可以配置為“，”逗號分割的表達式，上面的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=10.0.0.201:2181,10.0.0.202:2181,10.0.0.203:2181 #設置zookeeper的連接端口
上面是參數的解釋，實際的修改項為：
broker.id=201 每台服務器的broker.id都不能相同 listeners=PLAINTEXT://10.0.0.201:9092 #hostname host.name=10.0.0.201 # Log Basics log.dirs=/opt/kafka/logs #在log.retention.hours=168 下面新增下面三項 message.max.byte=5242880 default.replication.factor=2 replica.fetch.max.bytes=5242880 #設置zookeeper的連接端口 zookeeper.connect=10.0.0.201:2181,10.0.0.202:2181,10.0.0.203:2181
5、異常狀況
啟動kafka過一會進程自動掛掉問題原因
這是因為kafka logs目錄下的meta.properties文件中的broker.id與server.properties中的broker.id不一致所導致，只需把兩者改為一致啟動kafka后就不會自動掛掉了
kafka linux系統上使用時異常：
錯誤信息：
WARN Error while fetching metadata with correlation id xxx
異常截圖
解決辦法
修改config下的 server.properties 文件 將 listteners=PLAINTEXT://:9092 修改成listteners=PLAINTEXT://ip:9092
1、啟動服務
#從后台啟動Kafka集群（3台都需要啟動） /opt/kafka/kafka-server-start.sh -daemon /opt/kafka/config/server.properties
2、檢查服務是否啟動
#執行命令jps 20348 Jps 4233 QuorumPeerMain 18991 Kafka
4、啟動Kafka集群並測試
1、創建Topic來驗證是否創建成功
更多請看官方文檔：http://kafka.apache.org/documentation.html
#創建Topic /opt/kafka/bin/kafka-topics.sh --create --zookeeper 10.0.0.201:2181 --replication-factor 2 --partitions 1 --topic stock #解釋 --replication-factor 2 #復制兩份 --partitions 1 #創建1個分區 --topic #主題為stock '''在一台服務器上創建一個發布者''' #創建一個broker，發布者 /opt/kafka/bin/kafka-console-producer.sh --broker-list 10.0.0.201:9092,10.0.0.202:9092,10.0.0.203:9092 --topic stock '''在一台服務器上創建一個訂閱者''' /opt/kafka/bin/kafka-console-consumer.sh --zookeeper 10.0.0.201:2181,10.0.0.202:2181,10.0.0.203:2181 --topic stock --from-beginning
測試（在發布者那里發布消息看看訂閱者那里是否能正常收到~）：
4、其他命令
大部分命令可以去官方文檔查看
4.1、查看topic
/opt/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181
#就會顯示我們創建的所有topic
4.2、查看topic狀態
/kafka-topics.sh --describe --zookeeper localhost:2181 --topic stock #下面是顯示信息 Topic:ssports PartitionCount:1 ReplicationFactor:2 Configs: Topic: stock Partition: 0 Leader: 1 Replicas: 0,1 Isr: 1 #分區為為1 復制因子為2 stock的分區為0 #Replicas: 0,1 復制的為0，1
kafka集群搭建完畢
5、其他說明標注
5.1、日志說明
默認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.
5.2、上面的大家你完成之后可以登錄zk來查看zk的目錄情況
#使用客戶端進入zk ./zkCli.sh -server 127.0.0.1:2181 #默認是不用加’-server‘參數的因為我們修改了他的端口 #查看目錄情況 執行“ls /” [zk: 127.0.0.1:12181(CONNECTED) 0] ls / #顯示結果：[consumers, config, controller, isr_change_notification, admin, brokers, zookeeper, controller_epoch] ''' 上面的顯示結果中：只有zookeeper是，zookeeper原生的，其他都是Kafka創建的 ''' #標注一個重要的 [zk: 127.0.0.1:12181(CONNECTED) 1] get /brokers/ids/0 {"jmx_port":-1,"timestamp":"1456125963355","endpoints":["PLAINTEXT://10.0.0.201:9092"],"host":"192.168.7.100","version":2,"port":9092} cZxid = 0x1000001c1 ctime = Mon Feb 22 15:26:03 CST 2016 mZxid = 0x1000001c1 mtime = Mon Feb 22 15:26:03 CST 2016 pZxid = 0x1000001c1 cversion = 0 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x152e40aead20016 dataLength = 139 numChildren = 0 [zk: 127.0.0.1:2181(CONNECTED) 2] #還有一個是查看partion [zk: 127.0.0.1:2181(CONNECTED) 7] get /brokers/topics/shuaige/partitions/0 null cZxid = 0x100000029 ctime = Mon Feb 22 10:05:11 CST 2016 mZxid = 0x100000029 mtime = Mon Feb 22 10:05:11 CST 2016 pZxid = 0x10000002a cversion = 1 dataVersion = 0 aclVersion = 0 ephemeralOwner = 0x0 dataLength = 0 numChildren = 1 [zk: 127.0.0.1:2181(CONNECTED) 8]
Python 集群測試代碼
生產者
# encoding:utf-8 # Author:Richie # Date:2019/5/13 import json from kafka import KafkaProducer producer = KafkaProducer(value_serializer=lambda v: json.dumps(v).encode('utf-8'),bootstrap_servers=['10.0.0.201:9092','10.0.0.202:9092','10.0.0.203:9092']) msg_dict = { "sleep_time": 10, "db_config": { "database": "test_1", "host": "localhost", "user": "root", "password": "root" }, "table": "msg", "msg": "Hello World" } producer.send('stock', msg_dict, partition=0) producer.close()
消費者
# encoding:utf-8 # Author:Richie # Date:2019/5/13 from kafka import KafkaConsumer consumer = KafkaConsumer('stock', bootstrap_servers=['10.0.0.201:9092','10.0.0.202:9092','10.0.0.203:9092']) for msg in consumer: recv = "%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value) print(recv)