Ubuntu16.04安裝Kafka集羣

下載

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            每一個節點不一樣

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

複製到其它節點

注意要事先在zk中建立/kafka節點，不然會報出錯誤：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

[]

以守護進程方式啓動kafka

root@master:/usr/local/kafka_2.11-0.11.0.0# nohup bin/kafka-server-start.sh config/server.properties &

建立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

發送消息

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

查看集羣狀態信息

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/

生產服務器配置

# 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是一種高吞吐量的分佈式發佈訂閱的消息隊列系統，本來開發自LinkedIn，用做LinkedIn的活動流（ActivityStream）和運營數據處理管道（Pipeline）的基礎。如今它已被多家不一樣類型的公司做爲多種類型的數據管道和消息系統使用。

1 Kafka消息隊列簡介

1.1 基本術語

• Broker
  Kafka集羣包含一個或多個服務器，這種服務器被稱爲broker[5] 

• Topic
  每條發佈到Kafka集羣的消息都有一個類別，這個類別被稱爲Topic。（物理上不一樣Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存於一個或多個broker上但用戶只需指定消息的Topic便可生產或消費數據而沒必要關心數據存於何處）

• Partition
  Partition是物理上的概念，每一個Topic包含一個或多個Partition.（通常爲kafka節點數cpu的總核數）

• Producer
  負責發佈消息到Kafka broker

• Consumer
  消息消費者，向Kafka broker讀取消息的客戶端。

• Consumer Group
  每一個Consumer屬於一個特定的Consumer Group（可爲每一個Consumer指定group name，若不指定group name則屬於默認的group）。

1.2 消息隊列

1.2.1 基本特性

1. 可擴展

2. 在不須要下線的狀況下進行擴容

3. 數據流分區(partition)存儲在多個機器上

4. 高性能

5. 單個broker就能服務上千客戶端

6. 單個broker每秒種讀/寫可達每秒幾百兆字節

7. 多個brokers組成的集羣將達到很是強的吞吐能力

8. 性能穩定，不管數據多大

9. Kafka在底層摒棄了Java堆緩存機制，採用了操做系統級別的頁緩存，同時將隨機寫操做改成順序寫，再結合Zero-Copy的特性極大地改善了IO性能。

10. 持久存儲

11. 存儲在磁盤上

12. 冗餘備份到其餘服務器上以防止丟失

1.2.2 消息格式

1. 一個topic對應一種消息格式，所以消息用topic分類

2. 一個topic表明的消息有1個或者多個patition(s)組成

3. 一個partition中

4. 一個partition應該存放在一到多個server上

• 一個server爲leader

• 其餘servers爲followers

• leader須要接受讀寫請求

• followers僅做冗餘備份

• leader出現故障，會自動選舉一個follower做爲leader，保證服務不中斷

• 每一個server均可能扮演一些partitions的leader和其它partitions的follower角色，這樣整個集羣就會達到負載均衡的效果

• 若是隻有一個server，就沒有冗餘備份，是單機而不是集羣

• 若是有多個server

• 
  

消息按順序存放消息順序不可變只能追加消息，不能插入每一個消息都有一個offset，用做消息ID, 在一個partition中惟一offset有consumer保存和管理，所以讀取順序其實是徹底有consumer決定的，不必定是線性的消息有超時日期，過時則刪除

1.2.3 生產者 producer

• producer將消息寫入kafka

• 寫入要指定topic和partition

• 消息如何分到不一樣的partition，算法由producer指定

1.2.4 消費者 consumer

• consumer讀取消息並做處理

• consumer group

• 
  

  • 能夠按照partition的數目進行併發處理

  • 每一個partition都只有一個consumer讀取，於是保證了消息被處理的順序是按照partition的存放順序進行，注意這個順序受到producer存放消息的算法影響

  • 這個概念的引入爲了支持兩種場景：每條消息分發一個消費者，每條消息廣播給全部消費者

  • 多個consumer group訂閱一個topic，該topci的消息廣播給全部consumer group

  • 一條消息發送到一個consumer group後，只能由該group的一個consumer接收和使用

  • 一個group中的每一個consumer對應一個partition能夠帶來以下好處

  • 
    

•  一個Consumer能夠有多個線程進行消費，線程數應很少於topic的partition數，由於對於一個包含一或多消費線程的consumer group來講，一個partition只能分給其中的一個消費線程消費，且讓儘量多的線程能分配到partition（不過實際上真正去消費的線程及線程數仍是由線程池的調度機制來決定）。這樣若是線程數比partition數多，那麼單射分配也會有多出的線程，它們就不會消費到任何一個partition的數據而空轉耗資源 。

• 若是consumer從多個partition讀到數據，不保證數據間的順序性，kafka只保證在一個partition上數據是有序的，但多個partition，根據你讀的順序會有不一樣

• 增減consumer，broker，partition會致使rebalance，因此rebalance後consumer對應的partition會發生變化