1、kafka簡介
在流式計算中，Kafka一般用來緩存數(shù)據(jù)，Storm通過消費Kafka的數(shù)據(jù)進行計算冠王。
KAFKA + STORM +REDIS
1、Apache Kafka是一個開源消息系統(tǒng)秧骑，由Scala寫成。是由Apache軟件基金會開發(fā)的一個開源消息系統(tǒng)項目扣囊。
2乎折、Kafka最初是由LinkedIn開發(fā)，并于2011年初開源侵歇。2012年10月從Apache Incubator畢業(yè)骂澄。該項目的目標是為處理實時數(shù)據(jù)提供一個統(tǒng)一、高通量惕虑、低等待的平臺坟冲。
3、Kafka是一個分布式消息隊列：生產(chǎn)者溃蔫、消費者的功能健提。它提供了類似于JMS的特性，但是在設計實現(xiàn)上完全不同伟叛，此外它并不是JMS規(guī)范的實現(xiàn)私痹。
4、Kafka對消息保存時根據(jù)Topic進行歸類统刮，發(fā)送消息者稱為Producer,消息接受者稱為Consumer,此外kafka集群有多個kafka實例組成紊遵，每個實例(server)稱為broker。
5侥蒙、無論是kafka集群暗膜，還是producer和consumer都依賴于zookeeper集群保存一些meta信息，來保證系統(tǒng)可用性
注：類JMS消息隊列鞭衩，結合JMS中的兩種模式学搜，可以有多個消費者主動拉取數(shù)據(jù)，在JMS中只有點對點模式才有消費者主動拉取數(shù)據(jù)论衍。
2恒水、kafka集群部署
kafka集群跟storm集群類似，分為：下載安裝包饲齐、解壓安裝包钉凌、修改配置文件、分發(fā)安裝包（之后需要在各個機器上再次修改配置文件）捂人、啟動集群御雕。
準備工作
這里使用三臺機器進行集群:mini1,mini2,mini3
關閉防火墻矢沿，各機器的hosts文件如下
127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4
::1 localhost localhost.localdomain localhost6 localhost6.localdomain6 localhost.jinbm
192.168.25.127 mini1
192.168.25.129 mini2
192.168.25.130 mini3
1、下載安裝包并上傳到mini1機器上酸纲，這里下載的是 kafka_2.11-0.11.0.0.tgz捣鲸；
2、解壓安裝包導指定目錄闽坡，我這臺機器的apps目錄下安裝了很多大數(shù)據(jù)相關的軟件栽惶，所以解壓到該目錄了，并且重命名疾嗅。
[root@mini1 ~]# tar -zxvf kafka_2.11-0.11.0.0.tgz -C apps/
[root@mini1 ~]# cd apps/
[root@mini1 apps]# ll
總用量 28
drwxr-xr-x. 8 root root 4096 10月 19 15:15 apache-flume-1.6.0-bin
drwxrwxr-x. 10 hadoop hadoop 4096 9月 30 22:04 hadoop-2.6.4
drwxr-xr-x. 7 root root 4096 10月 30 00:20 hbase
drwxr-xr-x. 8 root root 4096 10月 17 12:38 hive
drwxr-xr-x. 5 root root 4096 1月 29 2015 kafka_2.11-0.11.0.0
drwxr-xr-x. 10 root root 4096 10月 30 15:34 storm
drwxr-xr-x. 10 root root 4096 10月 29 23:21 zookeeper-3.4.6
[root@mini1 apps]# mv kafka_2.11-0.11.0.0/ kafka
[root@mini1 apps]# ll
總用量 28
drwxr-xr-x. 8 root root 4096 10月 19 15:15 apache-flume-1.6.0-bin
drwxrwxr-x. 10 hadoop hadoop 4096 9月 30 22:04 hadoop-2.6.4
drwxr-xr-x. 7 root root 4096 10月 30 00:20 hbase
drwxr-xr-x. 8 root root 4096 10月 17 12:38 hive
drwxr-xr-x. 5 root root 4096 1月 29 2015 kafka
drwxr-xr-x. 10 root root 4096 10月 30 15:34 storm
drwxr-xr-x. 10 root root 4096 10月 29 23:21 zookeeper-3.4.6
3外厂、修改配置文件，進入kafka的config目錄下代承，可以查看到各類配置文件汁蝶，這里修改server.properties 。
[root@mini1 config]# ll
總用量 32
-rw-r--r--. 1 root root 1199 1月 29 2015 consumer.properties
-rw-r--r--. 1 root root 3846 1月 29 2015 log4j.properties
-rw-r--r--. 1 root root 2228 1月 29 2015 producer.properties
-rw-r--r--. 1 root root 5559 1月 29 2015 server.properties
-rw-r--r--. 1 root root 3325 1月 29 2015 test-log4j.properties
-rw-r--r--. 1 root root 993 1月 29 2015 tools-log4j.properties
-rw-r--r--. 1 root root 1023 1月 29 2015 zookeeper.properties
[root@mini1 config]# vi server.properties
#broker的全局唯一編號论悴，不能重復
broker.id=0
listeners=PLAINTEXT://mini1:9092
#處理網(wǎng)絡請求的線程數(shù)量
num.network.threads=3
#用來處理磁盤IO的線程數(shù)量
num.io.threads=8
#發(fā)送套接字的緩沖區(qū)大小
socket.send.buffer.bytes=102400
#接受套接字的緩沖區(qū)大小
socket.receive.buffer.bytes=102400
#請求套接字的緩沖區(qū)大小
socket.request.max.bytes=104857600
#kafka運行日志存放的路徑
log.dirs=/root/apps/kafka
#topic在當前broker上的分片個數(shù)
num.partitions=2
#用來恢復和清理data下數(shù)據(jù)的線程數(shù)量
num.recovery.threads.per.data.dir=1
#segment文件保留的最長時間掖棉，超時將被刪除
log.retention.hours=168
#滾動生成新的segment文件的最大時間
log.roll.hours=168
#日志文件中每個segment的大小，默認為1G
log.segment.bytes=1073741824
#周期性檢查文件大小的時間
log.retention.check.interval.ms=300000
#日志清理是否打開
log.cleaner.enable=true
#broker需要使用zookeeper保存meta數(shù)據(jù)
zookeeper.connect=mini1:2181,mini2:2181,mini3:2181
#zookeeper鏈接超時時間
zookeeper.connection.timeout.ms=6000
#partion buffer中膀估，消息的條數(shù)達到閾值幔亥，將觸發(fā)flush到磁盤
log.flush.interval.messages=10000
#消息buffer的時間，達到閾值察纯，將觸發(fā)flush到磁盤
log.flush.interval.ms=3000
#刪除topic需要server.properties中設置delete.topic.enable=true否則只是標記刪除
delete.topic.enable=true
4紫谷、將mini1上的kafka分發(fā)到mini2和mini3上
[root@mini1 ~]# scp -r apps/kafka/ mini2:/root/apps/
[root@mini1 ~]# scp -r apps/kafka/ mini3:/root/apps/
5、再次修改mini2和mini3上的配置文件server.properties捐寥，修改兩處笤昨，一個是broker編號一個是主機名
mini2上如下
#broker的全局唯一編號，不能重復
broker.id=1
listeners=PLAINTEXT://mini2:9092
mini3上如下
#broker的全局唯一編號握恳，不能重復
broker.id=2
listeners=PLAINTEXT://mini3:9092
6瞒窒、到這里配置完了，但是為了不進入kafka的bin目錄下就能執(zhí)行命令行操作乡洼，所以需要修改環(huán)境變量崇裁。三臺機器上修改均如下(添加最后兩行)
[root@mini1 ~]# vi /etc/profile
...
JAVA_HOME=/heima32/jdk1.7.0_55/
HADOOP_HOME=/root/apps/hadoop-2.6.4
export PATH=$JAVA_HOME/bin:$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
export HBASE_HOME=/root/apps/hbase
export PATH=$PATH:$HBASE_HOME/bin
export STORM_HOME=/root/apps/storm
export PATH=$PATH:$STORM_HOME/bin
export KAFKA_HOME=/root/apps/kafka
export PATH=$PATH:$KAFKA_HOME/bin
7、啟動集群
依次啟動mini1束昵，mini2和mini3上的kafka（進入kafka安裝目錄執(zhí)行以下命令）
[root@mini1 kafka]# bin/kafka-server-start.sh config/server.properties
3拔稳、kafka命令行操作
1、創(chuàng)建topic（注：因為修改了環(huán)境變量锹雏，所以不管有沒有進入bin目錄都能執(zhí)行相應命令行操作）
參數(shù)表示：指定副本數(shù)巴比，使用幾個partition（分片），topic名稱
[root@mini1 bin]# ./kafka-topics.sh --create --zookeeper mini1:2181 --replication-factor 2 --partitions 3 --topic orderMq
Created topic "orderMq".
[root@mini1 kafka]# kafka-topics.sh --create --zookeeper mini1:2181 --replication-factor 2 --partitions 3 --topic test
Created topic "test".
2、查看當前存在哪些topic
[root@mini1 kafka]# kafka-topics.sh --list --zookeeper mini1:2181
orderMq
test
3轻绞、刪除指定topic
[root@mini1 kafka]# kafka-topics.sh --delete --zookeeper mini1:2181 --topic test
Topic test is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
[root@mini1 kafka]# kafka-topics.sh --list --zookeeper mini1:2181
orderMq
4采记、通過shell命令行發(fā)送message(消息)
[root@mini1 bin]# kafka-console-producer.sh --broker-list mini1:9092 --topic orderMq
hello tom
hi jerry
spring ^H
hhaah
xixi
nini
5、通過shell命令行消費消息
[root@mini1 bin]# kafka-console-consumer.sh --zookeeper mini1:2181 -from-beginning --topic orderMq
hello tom
hi jerry
spring
hhaah
xixi
nini
注：這里的生成和消費是同順序的政勃，但有時候是不會順序一致的唧龄，只能保證消費一個partition里面的消息時是一致，消費多個partition（上面指定了消息被發(fā)送到了4個partition）里面的是不能保證順序完全一致奸远。
6既棺、查看某個topic的詳情
orderMq這個topic有用了三個partition（自己創(chuàng)建的時候指定的），各自有2個副本（也是自己創(chuàng)建的時候指定的）以及對于的leader（用處后面講）懒叛。
[root@mini2 bin]# ./kafka-topics.sh --topic orderMq --describe --zookeeper mini1:2181
Topic:orderMq PartitionCount:3 ReplicationFactor:2 Configs:
Topic: orderMq Partition: 0 Leader: 2 Replicas: 2,0 Isr: 0,2
Topic: orderMq Partition: 1 Leader: 0 Replicas: 0,1 Isr: 0,1
Topic: orderMq Partition: 2 Leader: 1 Replicas: 1,2 Isr: 1,2
接下來再創(chuàng)建個topic并且查看詳情（使用4個partition丸冕，兩個副本），并且進入運行日志存放的路徑查看
[root@mini2 bin]# kafka-topics.sh --create --zookeeper mini1:2181 --replication-factor 2 --partitions 4 --topic payment
Created topic "payment".
[root@mini2 bin]# kafka-topics.sh --topic payment --describe --zookeeper mini1:2181
Topic:payment PartitionCount:4 ReplicationFactor:2 Configs:
Topic: payment Partition: 0 Leader: 2 Replicas: 2,1 Isr: 2,1
Topic: payment Partition: 1 Leader: 0 Replicas: 0,2 Isr: 0,2
Topic: payment Partition: 2 Leader: 1 Replicas: 1,0 Isr: 1,0
Topic: payment Partition: 3 Leader: 2 Replicas: 2,0 Isr: 2,0
[root@mini1 kafka]# cd kafka-logs/
[root@mini1 kafka-logs]# ll
總用量 32
-rw-r--r--. 1 root root 4 11月 21 22:34 cleaner-offset-checkpoint
drwxr-xr-x. 2 root root 4096 11月 21 22:25 orderMq-0
drwxr-xr-x. 2 root root 4096 11月 21 22:25 orderMq-1
drwxr-xr-x. 2 root root 4096 11月 21 23:13 payment-1
drwxr-xr-x. 2 root root 4096 11月 21 23:13 payment-2
drwxr-xr-x. 2 root root 4096 11月 21 23:13 payment-3
-rw-r--r--. 1 root root 64 11月 22 03:38 recovery-point-offset-checkpoint
-rw-r--r--. 1 root root 64 11月 22 03:39 replication-offset-checkpoint
[root@mini1 kafka-logs]# cd orderMq-1
[root@mini1 orderMq-1]# ll
總用量 4
-rw-r--r--. 1 root root 10485760 11月 21 22:31 00000000000000000000.index
-rw-r--r--. 1 root root 194 11月 21 22:39 00000000000000000000.log
#00000000000000000000.log里面是發(fā)送的消息內容芍瑞，亂碼就沒貼出來了晨仑。
