zookeeper kafka集群
一、部署zookeeper集群
三臺機器上執行相同的操作
mkdir -p /data/zookeeper cd zookeeper-3.4.6 cp zoo_sample.cfg zoo.cfg [root@localhost conf]# more zoo_sample.cfg tickTime=2000 initLimit=10 syncLimit=5 dataDir=/data/zookeeper dataLogDir=/data/zookeeper/zkdatalog clientPort=2181 #maxClientCnxns=60 #autopurge.purgeInterval=1server.1=10.20.200.60:2888:3888 server.2=10.20.200.88:2888:3888 server.3=10.20.200.87:2888:3888
在dataDir目錄/data/zookeeper/下寫一個myid文件，命令如下：
echo 1 >myid
註意：這個id是zookeeper的主機標示，每個主機id不同，第二臺是2 第三臺是3，id需要與 zoo.cfg裏 server.數字相對應
啟動：
cd /opt/zookeeper-3.4.6/bin && sh zkServer.sh start
查看狀態：
sh zkServer.sh status
二、kafka集群安裝
啟動kafka至少需要jdk1.8 版本
首先安裝jdk1.8 tar包安裝
下載地址： http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html
mkdir /usr/local/java
將jdk 考到新創建的目錄下
設置環境變量
more /etc/profile JAVA_HOME=/usr/local/java/jdk1.8.0_131 JRE_HOME=/usr/local/java/jdk1.8.0_131/jre CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar:$JRE_HOME/lib PATH=$JAVA_HOME/bin:$PATH export PATH JAVA_HOME CLASSPATH
source /etc/profile
配置kafka集群
mkdir -p /data/kafka/kafkalogs tar xvf kafka_2.12-0.10.2.1.tgz mv kafka_2.12-0.10.2.1 /opt/kafak_2.12 cd /opt/kafak_2.12/config
主要配置server.properties
修改後配置文件如下：
broker.id=1 #每臺服務器的broker.id都不能相同 #hostname host.name=192.168.7.100 #在log.retention.hours=168 下面新增下面三項 message.max.byte=5242880 default.replication.factor=2 replica.fetch.max.bytes=5242880 #設置zookeeper的連接端口 zookeeper.connect=10.20.200.60:2181,10.20.200.87:2181,10.20.200.88:2181
配置文件說明：
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:2181,192.168.7.101:2181,192.168.7.107:2181 #設置zookeeper的連接端口
啟動：
bin/kafka-server-start.sh config/server.properties
後臺啟動
bin/kafka-server-start.sh -daemon config/server.properties
測試：
創建Topic來驗證是否創建成功
#創建Topic
./kafka-topics.sh --create --zookeeper 192.168.7.100:12181 --replication-factor 2 --partitions 1 --topic shuaige
#解釋
--replication-factor 2 #復制兩份
--partitions 1 #創建1個分區
--topic #主題為shuaige
在一臺服務器上創建一個發布者broker，發布者
./kafka-console-producer.sh --broker-list 192.168.7.100:19092 --topic shuaige
在一臺服務器上創建一個訂閱者
./kafka-console-consumer.sh --zookeeper localhost:12181 --topic shuaige --from-beginning
測試（在發布者那裏發布消息看看訂閱者那裏是否能正常收到~）
#查看topic
./kafka-topics.sh --list --zookeeper localhost:2181
4.2、查看topic狀態
/kafka-topics.sh --describe --zookeeper localhost:12181 --topic shuaige
#下面是顯示信息
Topic:ssports PartitionCount:1 ReplicationFactor:2 Configs:
Topic: shuaige Partition: 0 Leader: 1 Replicas: 0,1 Isr: 1
#分區為為1 復制因子為2 他的 shuaige的分區為0
#Replicas: 0,1 復制的為0，1
#
zookeeper kafka集群