下載
http://kafka.apache.org/downloads.html
http://mirror.bit.edu.cn/apache/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz
root@master:/usr/local/kafka_2.11-0.11.0.0/config# vim server.properties
broker.id=2 每個節(jié)點不同
log.retention.hours=168
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
zookeeper.connect=master:2181,slave1:2181,slave2:2181
復(fù)制到其它節(jié)點
注意要事先在zk中創(chuàng)建/kafka節(jié)點，否則會報出錯誤：java.lang.IllegalArgumentException: Path length must be > 0
root@master:/usr/local/zookeeper-3.4.9# bin/zkCli.sh -server master
[zk: master(CONNECTED) 7] create /kafka ''
Created /kafka
[zk: master(CONNECTED) 8] ls /
[cluster, controller, controller_epoch, brokers, zookeeper, kafka, admin, isr_change_notification, consumers, latest_producer_id_block, config]
[zk: master(CONNECTED) 9] ls /kafka
[]
以守護(hù)進(jìn)程方式啟動kafka
root@master:/usr/local/kafka_2.11-0.11.0.0# nohup bin/kafka-server-start.sh config/server.properties &
創(chuàng)建topic：
root@slave2:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh --create --zookeeper master:2181 --replication-factor 1 --partitions 1 --topic test
Created topic "test".
列出所有topic：
root@slave2:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh --list --zookeeper master:2181
test
發(fā)送消息
root@master:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-console-producer.sh --broker-list master:9092 --topic test
>this is a message
>this is ant^H message
消費消息
root@slave1:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-console-consumer.sh --zookeeper master:2181 --topic test --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
this is a message
this is an message
查看集群狀態(tài)信息
root@slave1:/usr/local/kafka_2.11-0.11.0.0# bin/kafka-topics.sh --describe --zookeeper slave1:2181 --topic my-replicated-topic
Topic:my-replicated-topic PartitionCount:1 ReplicationFactor:3 Configs:
Topic: my-replicated-topic Partition: 0 Leader: 3 Replicas: 1,3,2 Isr: 3,2
安裝kafka-manager
root@master:/usr/local/kafka_2.11-0.11.0.0# git clone https://github.com/yahoo/kafka-manager
root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager# cd kafka-manager/
root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager# ./sbt clean dist
[success] Total time: 3453 s, completed Aug 7, 2017 8:48:15 PM
打包好的文件存在
root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager/target/universal# ls
kafka-manager-1.3.3.12.zip tmp
修改kafka-manager配置文件
root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# vim conf/application.conf
kafka-manager.zkhosts="192.168.117.243:2181,192.168.117.45:2181,192.168.117.242:2181"
啟動kafka-manager
root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# bin/kafka-manager -Dconfig.file=conf/application.conf
推薦啟動方式：
root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=7778 &
登錄kafka manager:
http://192.168.117.243:7778/
root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# netstat -antlup | grep 7778
tcp6 0 0 :::7778 :::* LISTEN 100620/java
root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# bin/kafka-manager -Dconfig.file=conf/application.conf
This application is already running (Or delete /usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12/RUNNING_PID file).
停止kafka-manager
root@master:/usr/local/kafka_2.11-0.11.0.0/kafka-manager-1.3.3.12# rm RUNNING_PID
root@master:/usr/local/kafka_2.11-0.11.0.0# cd kafka-manager-1.0-SNAPSHOT/
生產(chǎn)服務(wù)器配置
# Replication configurations
num.replica.fetchers=4
replica.fetch.max.bytes=1048576
replica.fetch.wait.max.ms=500
replica.high.watermark.checkpoint.interval.ms=5000
replica.socket.timeout.ms=30000
replica.socket.receive.buffer.bytes=65536
replica.lag.time.max.ms=10000
replica.lag.max.messages=4000
controller.socket.timeout.ms=30000
controller.message.queue.size=10
# Log configuration
num.partitions=8
message.max.bytes=1000000
auto.create.topics.enable=true
log.index.interval.bytes=4096
log.index.size.max.bytes=10485760
log.retention.hours=168
log.flush.interval.ms=10000
log.flush.interval.messages=20000
log.flush.scheduler.interval.ms=2000
log.roll.hours=168
log.retention.check.interval.ms=300000
log.segment.bytes=1073741824
# ZK configuration
zookeeper.connection.timeout.ms=6000
zookeeper.sync.time.ms=2000
# Socket server configuration
num.io.threads=8
num.network.threads=8
socket.request.max.bytes=104857600
socket.receive.buffer.bytes=1048576
socket.send.buffer.bytes=1048576
queued.max.requests=16
fetch.purgatory.purge.interval.requests=100
producer.purgatory.purge.interval.requests=100
Kafka是一種高吞吐量的分布式發(fā)布訂閱的消息隊列系統(tǒng)，原本開發(fā)自LinkedIn，用作LinkedIn的活動流（ActivityStream）和運營數(shù)據(jù)處理管道（Pipeline）的基礎(chǔ)?，F(xiàn)在它已被多家不同類型的公司作為多種類型的數(shù)據(jù)管道和消息系統(tǒng)使用。
Broker
Kafka集群包含一個或多個服務(wù)器，這種服務(wù)器被稱為broker[5]
Topic
每條發(fā)布到Kafka集群的消息都有一個類別，這個類別被稱為Topic。（物理上不同Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存于一個或多個broker上但用戶只需指定消息的Topic即可生產(chǎn)或消費數(shù)據(jù)而不必關(guān)心數(shù)據(jù)存于何處）
Partition
Partition是物理上的概念，每個Topic包含一個或多個Partition.（一般為kafka節(jié)點數(shù)cpu的總核數(shù)）
Producer
負(fù)責(zé)發(fā)布消息到Kafka broker
Consumer
消息消費者，向Kafka broker讀取消息的客戶端。
Consumer Group
每個Consumer屬于一個特定的Consumer Group（可為每個Consumer指定group name，若不指定group name則屬于默認(rèn)的group）。
可擴(kuò)展
在不需要下線的情況下進(jìn)行擴(kuò)容
數(shù)據(jù)流分區(qū)(partition)存儲在多個機(jī)器上
高性能
單個broker就能服務(wù)上千客戶端
單個broker每秒種讀/寫可達(dá)每秒幾百兆字節(jié)
多個brokers組成的集群將達(dá)到非常強(qiáng)的吞吐能力
性能穩(wěn)定，無論數(shù)據(jù)多大
Kafka在底層摒棄了Java堆緩存機(jī)制，采用了操作系統(tǒng)級別的頁緩存，同時將隨機(jī)寫操作改為順序?qū)?，再結(jié)合Zero-Copy的特性極大地改善了IO性能。
持久存儲
存儲在磁盤上
冗余備份到其他服務(wù)器上以防止丟失
一個topic對應(yīng)一種消息格式，因此消息用topic分類
一個topic代表的消息有1個或者多個patition(s)組成
一個partition中
一個partition應(yīng)該存放在一到多個server上
一個server為leader
其他servers為followers
leader需要接受讀寫請求
followers僅作冗余備份
leader出現(xiàn)故障，會自動選舉一個follower作為leader，保證服務(wù)不中斷
每個server都可能扮演一些partitions的leader和其它partitions的follower角色，這樣整個集群就會達(dá)到負(fù)載均衡的效果
如果只有一個server，就沒有冗余備份，是單機(jī)而不是集群
如果有多個server
producer將消息寫入kafka
寫入要指定topic和partition
消息如何分到不同的partition，算法由producer指定
consumer讀取消息并作處理
consumer group
可以按照partition的數(shù)目進(jìn)行并發(fā)處理
每個partition都只有一個consumer讀取，因而保證了消息被處理的順序是按照partition的存放順序進(jìn)行，注意這個順序受到producer存放消息的算法影響
這個概念的引入為了支持兩種場景：每條消息分發(fā)一個消費者，每條消息廣播給所有消費者
多個consumer group訂閱一個topic，該topci的消息廣播給所有consumer group
一條消息發(fā)送到一個consumer group后，只能由該group的一個consumer接收和使用
一個group中的每個consumer對應(yīng)一個partition可以帶來如下好處
一個Consumer可以有多個線程進(jìn)行消費，線程數(shù)應(yīng)不多于topic的partition數(shù)，因為對于一個包含一或多消費線程的consumer group來說，一個partition只能分給其中的一個消費線程消費，且讓盡可能多的線程能分配到partition（不過實際上真正去消費的線程及線程數(shù)還是由線程池的調(diào)度機(jī)制來決定）。這樣如果線程數(shù)比partition數(shù)多，那么單射分配也會有多出的線程，它們就不會消費到任何一個partition的數(shù)據(jù)而空轉(zhuǎn)耗資源 。
如果consumer從多個partition讀到數(shù)據(jù)，不保證數(shù)據(jù)間的順序性，kafka只保證在一個partition上數(shù)據(jù)是有序的，但多個partition，根據(jù)你讀的順序會有不同
增減consumer，broker，partition會導(dǎo)致rebalance，所以rebalance后consumer對應(yīng)的partition會發(fā)生變化
另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。
重慶分公司
重慶分公司
