kafka_2.11-0.9.0.1.tgz 1.進入項目前的目錄 cd /home/dongshanxia mkdir kafka #創建項目目錄 cd kafka #進入項目目錄 mkdir kafkalogs #創建kafka消息目錄，主要存放kafka消息 2.進入配置文件目錄 cd /ho ...
1.進入項目前的目錄
cd /home/dongshanxia
mkdir kafka #創建項目目錄
cd kafka #進入項目目錄
mkdir kafkalogs #創建kafka消息目錄，主要存放kafka消息
2.進入配置文件目錄
cd /home/dongshanxia/kafka/kafka_2.11-0.9.0.1/config
//打開配置文件
vim server.properties
------------------------配置文件-------------------------------------
broker.id=1
/* 這是這台虛擬機上的值，在另外兩台虛擬機上應該是2或者3，這個值是唯一的，每台虛擬機或者叫伺服器不能相同。 */
listeners=PLAINTEXT://cjmaster01:9092
/設置本機IP和埠。我這裡設置的是listeners，
也可以直接設置host.name=cjmaster01,port=9092,
這個IP地址也是與本機相關的，每台伺服器上設置為自己的IP地址。/
log.dirs=/home/dongshanxia/kafka/kafkalogs
#指定其與另外幾台一樣的ip
zookeeper.connect=cjmaster01:2181,cjdatanode01:2181,cjdatanode02:2181
delete.topic.enable=true
~~~~~別人的配置文件
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的連接埠
命令集合
啟動kafka
bin/kafka-server-start.sh config/server.properties >/dev/null 2>&1 &
//守護模式-daemon
/home/dongshanxia/kafka/kafka_2.11-0.9.0.1/bin/kafka-server-start.sh -daemon /home/dongshanxia/kafka/kafka_2.11-0.9.0.1/config/server.properties
/home/dongshanxia/kafka/kafka_2.11-0.9.0.1/bin/kafka-server-start.sh /home/dongshanxia/kafka/kafka_2.11-0.9.0.1/config/server.properties >/dev/null 2>&1 &
創建TOPIC
創建一個叫做"test"的topic，它只有一個分區，一個副本
/home/dongshanxia/kafka/kafka_2.11-0.9.0.1/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
例如：
bin/kafka-topics.sh --zookeeper cjmaster01:2181 --create --topic mac-topic --replication-factor 2 --partitions 6
./bin/kafka-topics.sh --delete --zookeeper cjmaster01:2181 --topic 【topic name】
修改TOPIC分區數量
bin/kafka-topics.sh --zookeeper localhost:2182 --alter --partitions 20 --topic test
查看TOPIC
可以通過list命令查看創建的topic:
/home/dongshanxia/kafka/kafka_2.11-0.9.0.1/bin/kafka-topics.sh --list --zookeeper localhost:2181 test
查看topic的分區情況
/home/dongshanxia/kafka/kafka_2.11-0.9.0.1/bin/kafka-topics.sh --describe --zookeeper localhost:2181 test
消費數據
/home/dongshanxia/kafka/kafka_2.11-0.9.0.1/bin/kafka-console-consumer.sh --zookeeper 10.168.1.162:2181 --from-beginning --topic topic_hito_udp_gsm
/home/dongshanxia/kafka/kafka_2.11-0.9.0.1/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic topic_hito_tcp_mac
./kafka-console-consumer.sh --zookeeper localhost:2181 --topic topic_hik_udp_mac
//測試使用
/home/dongshanxia/kafka/kafka_2.11-0.9.0.1/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
/home/dongshanxia/kafka/kafka_2.11-0.9.0.1/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test
修改topic的備份數量
//校驗是否正確
bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ./topic-test.json --verify
輸出信息：修改之前報這個信息，說備份數量和文本不一致(因為文本內容是要添加副本數，肯定不一致），證明文本沒有問題，可以執行
[root@cjmaster01 kafka_2.11-0.9.0.1]# bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ./topic-test.json --verify
Status of partition reassignment:
ERROR: Assigned replicas (0) don't match the list of replicas for reassignment (0,1) for partition [topic-test,1]
ERROR: Assigned replicas (2) don't match the list of replicas for reassignment (1,2) for partition [topic-test,3]
ERROR: Assigned replicas (2) don't match the list of replicas for reassignment (2,1) for partition [topic-test,0]
ERROR: Assigned replicas (1) don't match the list of replicas for reassignment (1,2) for partition [topic-test,2]
ERROR: Assigned replicas (1) don't match the list of replicas for reassignment (1,0) for partition [topic-test,5]
ERROR: Assigned replicas (0) don't match the list of replicas for reassignment (1,0) for partition [topic-test,4]
Reassignment of partition [topic-test,1] failed
Reassignment of partition [topic-test,3] failed
Reassignment of partition [topic-test,0] failed
Reassignment of partition [topic-test,2] failed
Reassignment of partition [topic-test,5] failed
Reassignment of partition [topic-test,4] failed
//上邊返回成功了再執行
bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ./topic-test.json --execute
