kafka僞分布式安裝

必須先安裝zookeeper，參考：http://blog.csdn.net/u012572955/article/details/78910393

解壓檔案至 /usr/local/tanlei/kafkacluster/
tar -zxvf kafka_2.11-0.9.0.1.tgz


這裡隻建立了2個節點，複制兩個節點:
cp -r ./kafka_2.11-0.9.0.1/ ./kafka1
cp -r ./kafka_2.11-0.9.0.1/ ./kafka2


建立兩個節點的消息檔案夾，對應log.dirs=/usr/local/tanlei/kafkacluster/kafka2/kafkalogs
mkdir /usr/local/tanlei/kafkacluster/kafka1/kafkalogs
mkdir /usr/local/tanlei/kafkacluster/kafka2/kafkalogs


修改各自config下的server.properties檔案:
kafaka1的config下該檔案:
broker.id=0
port=9091
host.name=10.1.5.71
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=/usr/local/tanlei/kafkacluster/kafka1/kafkalogs
num.partitions=1
num.recovery.threads.per.data.dir=5
log.flush.interval.message=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.1.5.71:2181,10.1.5.71:2182,10.1.5.71:2183
zookeeper.connect.timeout.ms=6000


kafaka1的config下該檔案:
broker.id=1
port=9092
host.name=10.1.5.71
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=/usr/local/tanlei/kafkacluster/kafka2/kafkalogs
num.partitions=1
num.recovery.threads.per.data.dir=5
log.flush.interval.message=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.1.5.71:2181,10.1.5.71:2182,10.1.5.71:2183
zookeeper.connect.timeout.ms=6000


字段解釋:
broker.id=0         #目前機器在kafka機器裡唯一辨別，與zookeeper的myid一個意思，由于我使用獨立zookeeper這裡可以注釋掉
port=9092          #socket監聽的端口
host.name=192.168.152.192    #broker綁定的IP，我這裡三台機器都綁定了，後面遠端複制的時候記得修改
num.network.threads=3        #broker處理網絡的線程數目
num.io.threads=8                  #broker處理IO的線程數目
socket.send.buffer.bytes=102400     #kafka發送資料緩沖區的大小，到達門檻值才發送
socket.receive.buffer.bytes=102400     #kafka接收資料緩沖區的大小，達到門檻值才落地
socket.request.max.bytes=104857600    #kafka所能接受的資料最大值，100MB，這個值應該小于機器配置的Java heap，一般Java heap為記憶體的1/4
log.dirs=/home/kafka/kafkalogs        #存儲消息的目錄位置
num.partitions=1       #每個主題的預設分區數目，更多的分區可以允許更多的消費，可以提高性能
num.recovery.threads.per.data.dir        #在kafka啟動和關閉flush時，還原日志時，對于每個資料目錄所能使用的線程數目
log.flush.interval.message=10000        #當消息達到10000條時就會強制刷一次資料到磁盤
log.flush.interval.ms=1000                    #每過1秒就會強制刷資料到磁盤，機關毫秒
log.retention.hours=168                        #消息最大持久化時間，機關小時，即7天
log.retention.bytes=1073741824          #一種持久化政策，當剩餘段檔案不低于這個門檻值，目前段就會從日志消息中删除
log.segment.bytes=1073741824          #一個段檔案的最大值，達到最大值會另起一個新的segment
log.retention.check.interval.ms=300000   #每300秒去檢驗一下段檔案是否能被删除
zookeeper.connect=192.168.152.192:12181,192.168.152.193:12181,192.168.152.194:12181  #設定zookeeper的端口
zookeeper.connect.timeout.ms=6000    #連接配接zookeeper的逾時時間


分别啟動各節點:
進入bin目錄下，輸入指令行
./kafka-server-start.sh ../config/server.properties




建立啟動腳本:
/usr/local/tanlei/kafkacluster/kafka1/bin/kafka-server-start.sh /usr/local/tanlei/kafkacluster/kafka1/config/server.properties | /usr/local/tanlei/kafkacluster/kafka2/bin/kafka-server-start.sh /usr/local/tanlei/kafkacluster/kafka2/config/server.properties


建立topic:
./kafka-topics.sh --zookeeper 10.1.5.71:2181,10.1.5.71:2182,10.1.5.71:2183 --topic test1 --replication-factor 1 --partitions 1 --create
觀察topic:
./kafka-topics.sh --zookeeper 10.1.5.71:2181,10.1.5.71:2182,10.1.5.71:2183 --topic test1 --describe
           

kafka 文檔中文翻譯:

http://orchome.com/451