Kafka Producer APIs

舊版的Procuder API有兩種：kafka.producer.SyncProducer和kafka.producer.async.AsyncProducer.它們都實現了同一個介面：

1. class Producer {  
2.   /* 將訊息傳送到指定分割槽 */    
3.   public void send(kafka.javaapi.producer.ProducerData<K,V> producerData);  
4.   /* 批量傳送一批訊息 */    
5.   public void send(java.util.List<kafka.javaapi.producer.ProducerData<K,V>> producerData);  
6.   /* 關閉producer */    
7.   public void close();  
8. }  

新版的Producer API提供了以下功能：

1. 可以將多個訊息快取到本地佇列裡，然後非同步的批量傳送到broker，可以通過引數producer.type=async做到。快取的大小可以通過一些引數指定：queue.time和batch.size。一個後臺執行緒（(kafka.producer.async.ProducerSendThread）從佇列中取出資料並讓kafka.producer.EventHandler將訊息傳送到broker，也可以通過引數event.handler定製
   handler，在producer端處理資料的不同的階段註冊處理器，比如可以對這一過程進行日誌追蹤，或進行一些監控。只需實現kafka.producer.async.CallbackHandler介面，並在callback.handler中配置。
2. 自己編寫Encoder來序列化訊息，只需實現下面這個介面。預設的Encoder是kafka.serializer.DefaultEncoder。
   1. interface Encoder<T> {  
   2.   public Message toMessage(T data);  
   3. }  
3. 提供了基於Zookeeper的broker自動感知能力，可以通過引數zk.connect
   實現。如果不使用Zookeeper，也可以使用broker.list引數指定一個靜態的brokers列表，這樣訊息將被隨機的傳送到一個broker上，一旦選中的broker失敗了，訊息傳送也就失敗了。
4. 通過分割槽函式kafka.producer.Partitioner類對訊息分割槽。
   1. interface Partitioner<T> {  
   2.    int partition(T key, int numPartitions);  
   3. }  
   分割槽函式有兩個引數：key和可用的分割槽數量，從分割槽列表中選擇一個分割槽並返回id。預設的分割槽策略是hash(key)%numPartitions.如果key是null,就隨機的選擇一個。可以通過引數partitioner.class定製分割槽函式。

新的api完整例項如下：

1. import java.util.*;  
2. import kafka.javaapi.producer.Producer;  
3. import kafka.producer.KeyedMessage;  
4. import kafka.producer.ProducerConfig;  
5. public class TestProducer {  
6.     public static void main(String[] args) {  
7.         long events = Long.parseLong(args[0]);  
8.         Random rnd = new Random();  
9.         Properties props = new Properties();  
10.         props.put("metadata.broker.list", "broker1:9092,broker2:9092 ");  
11.         props.put("serializer.class", "kafka.serializer.StringEncoder");  
12.         props.put("partitioner.class", "example.producer.SimplePartitioner");  
13.         props.put("request.required.acks", "1");  
14.         ProducerConfig config = new ProducerConfig(props);  
15.         Producer<String, String> producer = new Producer<String, String>(config);  
16.         for (long nEvents = 0; nEvents < events; nEvents++) {   
17.                long runtime = new Date().getTime();    
18.                String ip = “192.168.2.” + rnd.nextInt(255);   
19.                String msg = runtime + “,www.example.com,” + ip;   
20.                KeyedMessage<String, String> data = new KeyedMessage<String, String>("page_visits", ip, msg);  
21.                producer.send(data);  
22.         }  
23.         producer.close();  
24.     }  
25. }  

下面這個是用到的分割槽函式：

1. import kafka.producer.Partitioner;  
2. import kafka.utils.VerifiableProperties;  
3. public class SimplePartitioner implements Partitioner<String> {  
4.     public SimplePartitioner (VerifiableProperties props) {  
5.     }  
6.     public int partition(String key, int a_numPartitions) {  
7.         int partition = 0;  
8.         int offset = key.lastIndexOf('.');  
9.         if (offset > 0) {  
10.            partition = Integer.parseInt( key.substring(offset+1)) % a_numPartitions;  
11.         }  
12.        return partition;  
13.   }  
14. }  

KafKa Consumer APIs

Consumer API有兩個級別。低級別的和一個指定的broker保持連線，並在接收完訊息後關閉連線，這個級別是無狀態的，每次讀取訊息都帶著offset。

高級別的API隱藏了和brokers連線的細節，在不必關心服務端架構的情況下和服務端通訊。還可以自己維護消費狀態，並可以通過一些條件指定訂閱特定的topic,比如白名單黑名單或者正則表示式。

低級別的API

1. class SimpleConsumer {  
2.   /*向一個broker傳送讀取請求並得到訊息集 */   
3.   public ByteBufferMessageSet fetch(FetchRequest request);  
4.   /*向一個broker傳送讀取請求並得到一個相應集 */   
5.   public MultiFetchResponse multifetch(List<FetchRequest> fetches);  
6.   /** 
7.    * 得到指定時間之前的offsets 
8.    * 返回值是offsets列表，以倒序排序 
9.    * @param time: 時間，毫秒, 
10.    *              如果指定為OffsetRequest$.MODULE$.LATIEST_TIME(), 得到最新的offset. 
11.    *              如果指定為OffsetRequest$.MODULE$.EARLIEST_TIME(),得到最老的offset. 
12.    */  
13.   public long[] getOffsetsBefore(String topic, int partition, long time, int maxNumOffsets);  
14. }  

低級別的API是高級別API實現的基礎，也是為了一些對維持消費狀態有特殊需求的場景，比如Hadoop consumer這樣的離線consumer。

高級別的API

1. /* 建立連線 */   
2. ConsumerConnector connector = Consumer.create(consumerConfig);  
3. interface ConsumerConnector {  
4.   /** 
5.    * 這個方法可以得到一個流的列表，每個流都是MessageAndMetadata的迭代，通過MessageAndMetadata可以拿到訊息和其他的元資料（目前之後topic）   
6.    *  Input: a map of <topic, #streams> 
7.    *  Output: a map of <topic, list of message streams> 
8.    */  
9.   public Map<String,List<KafkaStream>> createMessageStreams(Map<String,Int> topicCountMap);   
10.   /**     
11.    * 你也可以得到一個流的列表，它包含了符合TopicFiler的訊息的迭代， 
12.    * 一個TopicFilter是一個封裝了白名單或黑名單的正則表示式。 
13.    */  
14.   public List<KafkaStream> createMessageStreamsByFilter(  
15.       TopicFilter topicFilter, int numStreams);  
16.   /* 提交目前消費到的offset */  
17.   public commitOffsets()  
18.   /* 關閉連線 */  
19.   public shutdown()  
20. }  

這個API圍繞著由KafkaStream實現的迭代器展開，每個流代表一系列從一個或多個分割槽多和broker上匯聚來的訊息，每個流由一個執行緒處理，所以客戶端可以在建立的時候通過引數指定想要幾個流。一個流是多個分割槽多個broker的合併，但是每個分割槽的訊息只會流向一個流。

每呼叫一次createMessageStreams都會將consumer註冊到topic上，這樣consumer和brokers之間的負載均衡就會進行調整。API鼓勵每次呼叫建立更多的topic流以減少這種調整。createMessageStreamsByFilter方法註冊監聽可以感知新的符合filter的tipic。