三分鐘教你認識Kafka

1、基本概念

介紹

Kafka是一個分佈式的、可分區的、可複製的消息系統。它提供了普通消息系統的功能，但具備本身獨特的設計。

這個獨特的設計是什麼樣的呢？

首先讓咱們看幾個基本的消息系統術語：

Kafka將消息以topic爲單位進行概括。

將向Kafka topic發佈消息的程序成爲producers.

將預訂topics並消費消息的程序成爲consumer.

Kafka以集羣的方式運行，能夠由一個或多個服務組成，每一個服務叫作一個broker.

producers經過網絡將消息發送到Kafka集羣，集羣向消費者提供消息，以下圖所示：

客戶端和服務端經過TCP協議通訊。Kafka提供了Java客戶端，而且對多種語言都提供了支持。

Topics 和Logs

先來看一下Kafka提供的一個抽象概念:topic.

一個topic是對一組消息的概括。對每一個topic，Kafka 對它的日誌進行了分區，以下圖所示：

每一個分區都由一系列有序的、不可變的消息組成，這些消息被連續的追加到分區中。分區中的每一個消息都有一個連續的序列號叫作offset,用來在分區中惟一的標識這個消息。

在一個可配置的時間段內，Kafka集羣保留全部發布的消息，無論這些消息有沒有被消費。好比，若是消息的保存策略被設置爲2天，那麼在一個消息被髮布的兩天時間內，它都是能夠被消費的。以後它將被丟棄以釋放空間。Kafka的性能是和數據量無關的常量級的，因此保留太多的數據並非問題。

實際上每一個consumer惟一須要維護的數據是消息在日誌中的位置，也就是offset.這個offset有consumer來維護：通常狀況下隨着consumer不斷的讀取消息，這offset的值不斷增長，但其實consumer能夠以任意的順序讀取消息，好比它能夠將offset設置成爲一箇舊的值來重讀以前的消息。

以上特色的結合，使Kafka consumers很是的輕量級：它們能夠在不對集羣和其餘consumer形成影響的狀況下讀取消息。你可使用命令行來"tail"消息而不會對其餘正在消費消息的consumer形成影響。

將日誌分區能夠達到如下目的：首先這使得每一個日誌的數量不會太大，能夠在單個服務上保存。另外每一個分區能夠單獨發佈和消費，爲併發操做topic提供了一種可能。

分佈式

每一個分區在Kafka集羣的若干服務中都有副本，這樣這些持有副本的服務能夠共同處理數據和請求，副本數量是能夠配置的。副本使Kafka具有了容錯能力。

每一個分區都由一個服務器做爲「leader」，零或若干服務器做爲「followers」,leader負責處理消息的讀和寫，followers則去複製leader.若是leader down了，followers中的一臺則會自動成爲leader。集羣中的每一個服務都會同時扮演兩個角色：做爲它所持有的一部分分區的leader，同時做爲其餘分區的followers，這樣集羣就會據有較好的負載均衡。

Producers

Producer將消息發佈到它指定的topic中,並負責決定發佈到哪一個分區。一般簡單的由負載均衡機制隨機選擇分區，但也能夠經過特定的分區函數選擇分區。使用的更多的是第二種。

Consumers

發佈消息一般有兩種模式：隊列模式（queuing）和發佈-訂閱模式(publish-subscribe)。隊列模式中，consumers能夠同時從服務端讀取消息，每一個消息只被其中一個consumer讀到；發佈-訂閱模式中消息被廣播到全部的consumer中。Consumers能夠加入一個consumer 組，共同競爭一個topic，topic中的消息將被分發到組中的一個成員中。同一組中的consumer能夠在不一樣的程序中，也能夠在不一樣的機器上。若是全部的consumer都在一個組中，這就成爲了傳統的隊列模式，在各consumer中實現負載均衡。若是全部的consumer都不在不一樣的組中，這就成爲了發佈-訂閱模式，全部的消息都被分發到全部的consumer中。更常見的是，每一個topic都有若干數量的consumer組，每一個組都是一個邏輯上的「訂閱者」，爲了容錯和更好的穩定性，每一個組由若干consumer組成。這其實就是一個發佈-訂閱模式，只不過訂閱者是個組而不是單個consumer。

由兩個機器組成的集羣擁有4個分區 (P0-P3) 2個consumer組. A組有兩個consumerB組有4個

相比傳統的消息系統，Kafka能夠很好的保證有序性。

傳統的隊列在服務器上保存有序的消息，若是多個consumers同時從這個服務器消費消息，服務器就會以消息存儲的順序向consumer分發消息。雖然服務器按順序發佈消息，可是消息是被異步的分發到各consumer上，因此當消息到達時可能已經失去了原來的順序，這意味着併發消費將致使順序錯亂。爲了不故障，這樣的消息系統一般使用「專用consumer」的概念，其實就是隻容許一個消費者消費消息，固然這就意味着失去了併發性。

在這方面Kafka作的更好，經過分區的概念，Kafka能夠在多個consumer組併發的狀況下提供較好的有序性和負載均衡。將每一個分區分只分發給一個consumer組，這樣一個分區就只被這個組的一個consumer消費，就能夠順序的消費這個分區的消息。由於有多個分區，依然能夠在多個consumer組之間進行負載均衡。注意consumer組的數量不能多於分區的數量，也就是有多少分區就容許多少併發消費。

Kafka只能保證一個分區以內消息的有序性，在不一樣的分區之間是不能夠的，這已經能夠知足大部分應用的需求。若是須要topic中全部消息的有序性，那就只能讓這個topic只有一個分區，固然也就只有一個consumer組消費它。

2、環境搭建

Step 1: 下載Kafka

點擊下載最新的版本並解壓.

> tar -xzf kafka_2.9.2-0.8.1.1.tgz

> cd kafka_2.9.2-0.8.1.1

複製代碼

Step 2: 啓動服務

Kafka用到了Zookeeper，全部首先啓動Zookper，下面簡單的啓用一個單實例的Zookkeeper服務。能夠在命令的結尾加個&符號，這樣就能夠啓動後離開控制檯。

> bin/zookeeper-server-start.sh config/zookeeper.properties &

[2013-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig)

...

複製代碼

如今啓動Kafka:

> bin/kafka-server-start.sh config/server.properties

[2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties)

[2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties)

...

複製代碼

Step 3: 建立 topic

建立一個叫作「test」的topic，它只有一個分區，一個副本。

> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

複製代碼

能夠經過list命令查看建立的topic:

> bin/kafka-topics.sh --list --zookeeper localhost:2181

test

複製代碼

除了手動建立topic，還能夠配置broker讓它自動建立topic.

Step 4:發送消息.

Kafka 使用一個簡單的命令行producer，從文件中或者從標準輸入中讀取消息併發送到服務端。默認的每條命令將發送一條消息。

運行producer並在控制檯中輸一些消息，這些消息將被髮送到服務端：

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

This is a messageThis is another message

複製代碼

ctrl+c能夠退出發送。

Step 5: 啓動consumer

Kafka also has a command line consumer that will dump out messages to standard output.

Kafka也有一個命令行consumer能夠讀取消息並輸出到標準輸出：

> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

This is a message

This is another message

複製代碼

你在一個終端中運行consumer命令行，另外一個終端中運行producer命令行，就能夠在一個終端輸入消息，另外一個終端讀取消息。

這兩個命令都有本身的可選參數，能夠在運行的時候不加任何參數能夠看到幫助信息。

Step 6: 搭建一個多個broker的集羣

剛纔只是啓動了單個broker，如今啓動有3個broker組成的集羣，這些broker節點也都是在本機上的：

首先爲每一個節點編寫配置文件：

> cp config/server.properties config/server-1.properties

> cp config/server.properties config/server-2.properties

複製代碼

在拷貝出的新文件中添加如下參數：

config/server-1.properties:

broker.id=1

port=9093

log.dir=/tmp/kafka-logs-1

複製代碼

config/server-2.properties:

broker.id=2

port=9094

log.dir=/tmp/kafka-logs-2

複製代碼

broker.id在集羣中惟一的標註一個節點，由於在同一個機器上，因此必須制定不一樣的端口和日誌文件，避免數據被覆蓋。

We already have Zookeeper and our single node started, so we just need to start the two new nodes:

剛纔已經啓動可Zookeeper和一個節點，如今啓動另外兩個節點：

> bin/kafka-server-start.sh config/server-1.properties &

...

> bin/kafka-server-start.sh config/server-2.properties &

...

複製代碼

建立一個擁有3個副本的topic:

> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

複製代碼

如今咱們搭建了一個集羣，怎麼知道每一個節點的信息呢？運行「"describe topics」命令就能夠了：

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic

複製代碼

Topic:my-replicated-topic       PartitionCount:1        ReplicationFactor:3     Configs:

Topic: my-replicated-topic      Partition: 0    Leader: 1       Replicas: 1,2,0 Isr: 1,2,0

複製代碼

下面解釋一下這些輸出。第一行是對全部分區的一個描述，而後每一個分區都會對應一行，由於咱們只有一個分區因此下面就只加了一行。

leader：負責處理消息的讀和寫，leader是從全部節點中隨機選擇的.

replicas：列出了全部的副本節點，無論節點是否在服務中.

isr：是正在服務中的節點.

在咱們的例子中，節點1是做爲leader運行。

向topic發送消息：

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic

複製代碼

...

my test message 1my test message 2^C

複製代碼

消費這些消息：

> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic

...

my test message 1

my test message 2

^C

測試一下容錯能力.Broker 1做爲leader運行，如今咱們kill掉它：

> ps | grep server-1.properties7564 ttys002    0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.6/Home/bin/java...

> kill -9 7564

複製代碼

另一個節點被選作了leader,node 1 再也不出如今 in-sync 副本列表中：

> bin/kafka-topics.sh --describe --zookeeper localhost:218192 --topic my-replicated-topic

Topic:my-replicated-topic       PartitionCount:1        ReplicationFactor:3     Configs:

Topic: my-replicated-topic      Partition: 0    Leader: 2       Replicas: 1,2,0 Isr: 2,0

複製代碼

雖然最初負責續寫消息的leader down掉了，但以前的消息仍是能夠消費的：

> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic

...

my test message 1

my test message 2

複製代碼

看來Kafka的容錯機制仍是不錯的。

 

3、搭建Kafka開發環境

咱們搭建了kafka的服務器，並可使用Kafka的命令行工具建立topic，發送和接收消息。下面咱們來搭建kafka的開發環境。

添加依賴

搭建開發環境須要引入kafka的jar包，一種方式是將Kafka安裝包中lib下的jar包加入到項目的classpath中，這種比較簡單了。不過咱們使用另外一種更加流行的方式：使用maven管理jar包依賴。

建立好maven項目後，在pom.xml中添加如下依賴：

org.apache.kafka

kafka_2.10

0.8.0

複製代碼

添加依賴後你會發現有兩個jar包的依賴找不到。不要緊我都幫你想好了，點擊這裏下載這兩個jar包，解壓後你有兩種選擇，第一種是使用mvn的install命令將jar包安裝到本地倉庫，另外一種是直接將解壓後的文件夾拷貝到mvn本地倉庫的com文件夾下，好比個人本地倉庫是d:\mvn,完成後個人目錄結構是這樣的：

配置程序

首先是一個充當配置文件做用的接口,配置了Kafka的各類鏈接參數：

package com.sohu.kafkademon;

public interface KafkaProperties

{

final static String zkConnect = "10.22.10.139:2181";

final static String groupId = "group1";

final static String topic = "topic1";

final static String kafkaServerURL = "10.22.10.139";

final static int kafkaServerPort = 9092;

final static int kafkaProducerBufferSize = 64 * 1024;

final static int connectionTimeOut = 20000;

final static int reconnectInterval = 10000;

final static String topic2 = "topic2";

final static String topic3 = "topic3";

final static String clientId = "SimpleConsumerDemoClient";

}

複製代碼

producer

package com.sohu.kafkademon;

import java.util.Properties;

import kafka.producer.KeyedMessage;

import kafka.producer.ProducerConfig;

/**

* @author leicui bourne_cui@163.com

*/

public class KafkaProducer extends Thread

{

private final kafka.javaapi.producer.Producer producer;

private final String topic;

private final Properties props = new Properties();

public KafkaProducer(String topic)

{

props.put("serializer.class", "kafka.serializer.StringEncoder");

props.put("metadata.broker.list", "10.22.10.139:9092");

producer = new kafka.javaapi.producer.Producer(new ProducerConfig(props));

this.topic = topic;

}

@Override

public void run() {

int messageNo = 1;

while (true)

{

String messageStr = new String("Message_" + messageNo);

System.out.println("Send:" + messageStr);

producer.send(new KeyedMessage(topic, messageStr));

messageNo++;

try {

sleep(3000);

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

}

}

}

複製代碼

consumer

package com.sohu.kafkademon;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Properties;

import kafka.consumer.ConsumerConfig;

import kafka.consumer.ConsumerIterator;

import kafka.consumer.KafkaStream;

import kafka.javaapi.consumer.ConsumerConnector;

/**

* @author leicui bourne_cui@163.com

*/

public class KafkaConsumer extends Thread

{

private final ConsumerConnector consumer;

private final String topic;

public KafkaConsumer(String topic)

{

consumer = kafka.consumer.Consumer.createJavaConsumerConnector(

createConsumerConfig());

this.topic = topic;

}

private static ConsumerConfig createConsumerConfig()

{

Properties props = new Properties();

props.put("zookeeper.connect", KafkaProperties.zkConnect);

props.put("group.id", KafkaProperties.groupId);

props.put("zookeeper.session.timeout.ms", "40000");

props.put("zookeeper.sync.time.ms", "200");

props.put("auto.commit.interval.ms", "1000");

return new ConsumerConfig(props);

}

@Override

public void run() {

Map topicCountMap = new HashMap();

topicCountMap.put(topic, new Integer(1));

Map>> consumerMap = consumer.createMessageStreams(topicCountMap);

KafkaStream stream = consumerMap.get(topic).get(0);

ConsumerIterator it = stream.iterator();

while (it.hasNext()) {

System.out.println("receive：" + new String(it.next().message()));

try {

sleep(3000);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}

複製代碼

簡單的發送接收

運行下面這個程序，就能夠進行簡單的發送接收消息了：

package com.sohu.kafkademon;

/**

* @author leicui bourne_cui@163.com

*/

public class KafkaConsumerProducerDemo

{

public static void main(String[] args)

{

KafkaProducer producerThread = new KafkaProducer(KafkaProperties.topic);

producerThread.start();

KafkaConsumer consumerThread = new KafkaConsumer(KafkaProperties.topic);

consumerThread.start();

}

}

複製代碼

高級別的consumer

下面是比較負載的發送接收的程序：

package com.sohu.kafkademon;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Properties;

import kafka.consumer.ConsumerConfig;

import kafka.consumer.ConsumerIterator;

import kafka.consumer.KafkaStream;

import kafka.javaapi.consumer.ConsumerConnector;

/**

* @author leicui bourne_cui@163.com

*/

public class KafkaConsumer extends Thread

{

private final ConsumerConnector consumer;

private final String topic;

public KafkaConsumer(String topic)

{

consumer = kafka.consumer.Consumer.createJavaConsumerConnector(

createConsumerConfig());

this.topic = topic;

}

private static ConsumerConfig createConsumerConfig()

{

Properties props = new Properties();

props.put("zookeeper.connect", KafkaProperties.zkConnect);

props.put("group.id", KafkaProperties.groupId);

props.put("zookeeper.session.timeout.ms", "40000");

props.put("zookeeper.sync.time.ms", "200");

props.put("auto.commit.interval.ms", "1000");

return new ConsumerConfig(props);

}

@Override

public void run() {

Map topicCountMap = new HashMap();

topicCountMap.put(topic, new Integer(1));

Map>> consumerMap = consumer.createMessageStreams(topicCountMap);

KafkaStream stream = consumerMap.get(topic).get(0);

ConsumerIterator it = stream.iterator();

while (it.hasNext()) {

System.out.println("receive：" + new String(it.next().message()));

try {

sleep(3000);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}

4、數據持久化

不要畏懼文件系統!

Kafka大量依賴文件系統去存儲和緩存消息。對於硬盤有個傳統的觀念是硬盤老是很慢，這使不少人懷疑基於文件系統的架構可否提供優異的性能。實際上硬盤的快慢徹底取決於使用它的方式。設計良好的硬盤架構能夠和內存同樣快。

在6塊7200轉的SATA RAID-5磁盤陣列的線性寫速度差很少是600MB/s，可是隨即寫的速度倒是100k/s，差了差很少6000倍。現代的操做系統都對次作了大量的優化，使用了 read-ahead 和 write-behind的技巧，讀取的時候成塊的預讀取數據，寫的時候將各類微小瑣碎的邏輯寫入組織合併成一次較大的物理寫入。對此的深刻討論能夠查看這裏，它們發現線性的訪問磁盤，不少時候比隨機的內存訪問快得多。

爲了提升性能，現代操做系統每每使用內存做爲磁盤的緩存，現代操做系統樂於把全部空閒內存用做磁盤緩存，雖然這可能在緩存回收和從新分配時犧牲一些性能。全部的磁盤讀寫操做都會通過這個緩存，這不太可能被繞開除非直接使用I/O。因此雖然每一個程序都在本身的線程裏只緩存了一份數據，但在操做系統的緩存裏還有一份，這等於存了兩份數據。

另外再來討論一下JVM,如下兩個事實是衆所周知的：

•Java對象佔用空間是很是大的，差很少是要存儲的數據的兩倍甚至更高。

•隨着堆中數據量的增長，垃圾回收回變的愈來愈困難。

基於以上分析，若是把數據緩存在內存裏，由於須要存儲兩份，不得不使用兩倍的內存空間，Kafka基於JVM，又不得不將空間再次加倍,再加上要避免GC帶來的性能影響，在一個32G內存的機器上，不得不使用到28-30G的內存空間。而且當系統重啓的時候，又必需要將數據刷到內存中（ 10GB 內存差很少要用10分鐘），就算使用冷刷新（不是一次性刷進內存，而是在使用數據的時候沒有就刷到內存）也會致使最初的時候新能很是慢。可是使用文件系統，即便系統重啓了，也不須要刷新數據。使用文件系統也簡化了維護數據一致性的邏輯。

因此與傳統的將數據緩存在內存中而後刷到硬盤的設計不一樣，Kafka直接將數據寫到了文件系統的日誌中。

常量時間的操做效率

在大多數的消息系統中，數據持久化的機制每每是爲每一個cosumer提供一個B樹或者其餘的隨機讀寫的數據結構。B樹固然是很棒的，可是也帶了一些代價：好比B樹的複雜度是O(log N)，O(log N)一般被認爲就是常量複雜度了，但對於硬盤操做來講並不是如此。磁盤進行一次搜索須要10ms，每一個硬盤在同一時間只能進行一次搜索，這樣併發處理就成了問題。雖然存儲系統使用緩存進行了大量優化，可是對於樹結構的性能的觀察結果卻代表，它的性能每每隨着數據的增加而線性降低，數據增加一倍，速度就會下降一倍。

直觀的講，對於主要用於日誌處理的消息系統，數據的持久化能夠簡單的經過將數據追加到文件中實現，讀的時候從文件中讀就行了。這樣作的好處是讀和寫都是 O(1) 的，而且讀操做不會阻塞寫操做和其餘操做。這樣帶來的性能優點是很明顯的，由於性能和數據的大小沒有關係了。

既然可使用幾乎沒有容量限制（相對於內存來講）的硬盤空間創建消息系統，就能夠在沒有性能損失的狀況下提供一些通常消息系統不具有的特性。好比，通常的消息系統都是在消息被消費後當即刪除，Kafka卻能夠將消息保存一段時間（好比一星期），這給consumer提供了很好的機動性和靈活性，這點在從此的文章中會有詳述。

5、消息傳輸的事務定義

以前討論了consumer和producer是怎麼工做的，如今來討論一下數據傳輸方面。數據傳輸的事務定義一般有如下三種級別：

最多一次: 消息不會被重複發送，最多被傳輸一次，但也有可能一次不傳輸。

最少一次: 消息不會被漏發送，最少被傳輸一次，但也有可能被重複傳輸.

精確的一次（Exactly once）: 不會漏傳輸也不會重複傳輸,每一個消息都傳輸被一次並且僅僅被傳輸一次，這是你們所指望的。

大多數消息系統聲稱能夠作到「精確的一次」，可是仔細閱讀它們的的文檔能夠看到裏面存在誤導，好比沒有說明當consumer或producer失敗時怎麼樣，或者當有多個consumer並行時怎麼樣，或寫入硬盤的數據丟失時又會怎麼樣。kafka的作法要更先進一些。當發佈消息時，Kafka有一個「committed」的概念，一旦消息被提交了，只要消息被寫入的分區的所在的副本broker是活動的，數據就不會丟失。關於副本的活動的概念，下節文檔會討論。如今假設broker是不會down的。

若是producer發佈消息時發生了網絡錯誤，但又不肯定實在提交以前發生的仍是提交以後發生的，這種狀況雖然不常見，可是必須考慮進去，如今Kafka版本尚未解決這個問題，未來的版本正在努力嘗試解決。

並非全部的狀況都須要「精確的一次」這樣高的級別，Kafka容許producer靈活的指定級別。好比producer能夠指定必須等待消息被提交的通知，或者徹底的異步發送消息而不等待任何通知，或者僅僅等待leader聲明它拿到了消息（followers沒有必要）。

如今從consumer的方面考慮這個問題，全部的副本都有相同的日誌文件和相同的offset，consumer維護本身消費的消息的offset，若是consumer不會崩潰固然能夠在內存中保存這個值，固然誰也不能保證這點。若是consumer崩潰了，會有另一個consumer接着消費消息，它須要從一個合適的offset繼續處理。這種狀況下能夠有如下選擇：

consumer能夠先讀取消息，而後將offset寫入日誌文件中，而後再處理消息。這存在一種可能就是在存儲offset後還沒處理消息就crash了，新的consumer繼續從這個offset處理，那麼就會有些消息永遠不會被處理，這就是上面說的「最多一次」。

consumer能夠先讀取消息，處理消息，最後記錄offset，固然若是在記錄offset以前就crash了，新的consumer會重複的消費一些消息，這就是上面說的「最少一次」。

「精確一次」能夠經過將提交分爲兩個階段來解決：保存了offset後提交一次，消息處理成功以後再提交一次。可是還有個更簡單的作法：將消息的offset和消息被處理後的結果保存在一塊兒。好比用Hadoop ETL處理消息時，將處理後的結果和offset同時保存在HDFS中，這樣就能保證消息和offser同時被處理了。

完整的項目源碼來源  歡迎你們一塊兒學習研究相關技術，源碼獲取請加求求:2670716182