kafka算是很麻煩的一件事兒，起因是最近需要採集大量的資料，原先是隻用了典型的high-level Consumer的API，最經典的不過如下：

1. Properties props = new Properties();      
2.        props.put("zookeeper.connect", "xxxx:2181");      
3.        props.put("zookeeper.connectiontimeout.ms", "1000000");      
4.        props.put("group.id", "test_group");    
5.        props.put("zookeeper.session.timeout.ms", "40000");    
6.        props.put("zookeeper.sync.time.ms", "200");    
7.        props.put("auto.commit.interval.ms", "1000");    
8.        ConsumerConfig consumerConfig = new ConsumerConfig(props);      
9.        ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);      
10.        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();      
11.        topicCountMap.put("test", new Integer(1));      
12.        //key--topic    
13.        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap =     
14.                consumerConnector.createMessageStreams(topicCountMap);    
15.        KafkaStream<byte[], byte[]> stream = consumerMap.get("test").get(0);    
16.        ConsumerIterator<byte[], byte[]> it = stream.iterator();    
17.        StringBuffer sb = new StringBuffer();    
18.        while(it.hasNext()) {    
19.            try {    
20.                String msg = new String(it.next().message(), "utf-8").trim();    
21.                System.out.println("receive:" + msg);    
22.            } catch (UnsupportedEncodingException e) {    
23.                e.printStackTrace();    
24.            }    
25.        }    

 這是典型的kafka消費端消費資料的程式碼，但可以看出這是十分典型的單執行緒消費。在本地玩玩熟悉kafka還行，（就跟入門Java學會寫main方法列印hello world一樣~~~~），問題是學的東西必須真正應用到實際中，你不可能只在單執行緒採集裡原地打轉吧。。so，多執行緒採集迫在眉急啊！！

本人研究卡夫卡多執行緒消費還是耗了一段時間的，希望把過程儘可能完整地記錄下來，以便各位同行有需要可以參考。。

首先，最好理解kafka的基本原理和一些基本概念，閱讀官網文件很有必要，這樣才會有一個比較清晰的概念，而不是跟無頭蒼蠅一樣亂撞——出了錯去網上查是灰常痛苦滴！！

http://kafka.apache.org/documentation.html

好了，大概說下卡夫卡的“分割槽·”的概念吧：

 這張圖比較清晰地描述了“分割槽”的概念，對於某一個topic的訊息來說，我們可以把這組訊息傳送給若干個分割槽，就相當於一組訊息分發一樣。

分割槽、Offset、消費執行緒、group.id的關係

1）一組（類）訊息通常由某個topic來歸類，我們可以把這組訊息“分發”給若干個分割槽（partition），每個分割槽的訊息各不相同；

2）每個分割槽都維護著他自己的偏移量（Offset），記錄著該分割槽的訊息此時被消費的位置；

3）一個消費執行緒可以對應若干個分割槽，但一個分割槽只能被具體某一個消費執行緒消費；

4）group.id用於標記某一個消費組，每一個消費組都會被記錄他在某一個分割槽的Offset，即不同consumer group針對同一個分割槽，都有“各自”的偏移量。

說完概念，必須要注意的一點是，必須確認卡夫卡的server.properties裡面的一個屬性num.partitions必須被設定成大於1的值，否則消費端再怎麼折騰，也用不了多執行緒哦。我這裡的環境下，該屬性值被設定成10了。

重構一下上述經典的消費端程式碼：

1. publicclass KafakConsumer implements Runnable {  
2.     private ConsumerConfig consumerConfig;  
3.     privatestatic String topic="blog";  
4.     Properties props;  
5.     finalint a_numThreads = 6;  
6.     public KafakConsumer() {  
7.         props = new Properties();      
8.         props.put("zookeeper.connect", "xxx:2181,yyy:2181,zzz:2181");    
9. //      props.put("zookeeper.connect", "localhost:2181");    
10. //        props.put("zookeeper.connectiontimeout.ms", "30000");    
11.         props.put("group.id", "blog");    
12.         props.put("zookeeper.session.timeout.ms", "400");  
13.         props.put("zookeeper.sync.time.ms", "200");  
14.         props.put("auto.commit.interval.ms", "1000");   
15.         props.put("auto.offset.reset", "smallest");  
16.         consumerConfig = new ConsumerConfig(props);  
17.     }  
18.     @Override
19.     publicvoid run() {  
20.         Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
21.         topicCountMap.put(topic, new Integer(a_numThreads));  
22.         ConsumerConfig consumerConfig = new ConsumerConfig(props);  
23.         ConsumerConnector consumer = kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);  
24.         Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);  
25.         List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);  
26.         ExecutorService executor = Executors.newFixedThreadPool(a_numThreads);  
27.         for (final KafkaStream stream : streams) {  
28.             executor.submit(new KafkaConsumerThread(stream));  
29.         }  
30.     }  
31.     publicstaticvoid main(String[] args) {  
32.         System.out.println(topic);  
33.         Thread t = new Thread(new KafakConsumer());  
34.         t.start();  
35.     }  
36. }  

 從這段重構的程式碼可以看出，KafkaStream<byte[], byte[]> stream = consumerMap.get("test").get(0); 這行程式碼已被廢棄，得到List<KafkaStream<byte[], byte[]>>之後不再是得到他的頭元素get(0)就ok了，而且topicCountMap.put(topic, new Integer(a_numThreads));的第二個引數也不再是1.

 其中，具體消費執行緒KafkaConsumerThread程式碼為：

1. publicclass KafkaConsumerThread implements Runnable {  
2.     private KafkaStream<byte[], byte[]> stream;  
3.     public KafkaConsumerThread(KafkaStream<byte[], byte[]> stream) {  
4.         this.stream = stream;  
5.     }  
6.     @Override
7.     publicvoid run() {  
8.         ConsumerIterator<byte[], byte[]> it = stream.iterator();  
9.         while (it.hasNext()) {  
10.             MessageAndMetadata<byte[], byte[]> mam = it.next();  
11.             System.out.println(Thread.currentThread().getName() + ": partition[" + mam.partition() + "],"
12.             + "offset[" + mam.offset() + "], " + new String(mam.message()));  
13.         }  
14.     }  
15. }  

相關推薦