4. 使用Kafka的Producer API來完成訊息的推送

1) Kafka 0.9.0.1的java client依賴：

Xml程式碼 

1. <dependency>
2.     <groupId>org.apache.kafka</groupId>
3.     <artifactId>kafka-clients</artifactId>
4.     <version>0.9.0.1</version>
5. </dependency>

2) 寫一個KafkaUtil工具類，用於構造Kafka Client

Java程式碼 

1. publicclass KafkaUtil {  
2.     privatestatic KafkaProducer<String, String> kp;  
3.     publicstatic KafkaProducer<String, String> getProducer() {  
4.         if (kp == null) {  
5.             Properties props = new Properties();  
6.             props.put("bootstrap.servers", "10.0.0.100:9092,10.0.0.101:9092"
   );  
7.             props.put("acks", "1");  
8.             props.put("retries", 0);  
9.             props.put("batch.size", 16384);  
10.             props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
11.             props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"
    );  
12.             kp = new KafkaProducer<String, String>(props);  
13.         }  
14.         return kp;  
15.     }  
16. }  

  KafkaProducer<K,V>的K代表每條訊息的key型別，V代表訊息型別。訊息的key用於決定此條訊息由哪一個partition接收，所以我們需要保證每條訊息的key是不同的。

  Producer端的常用配置

• bootstrap.servers：Kafka叢集連線串，可以由多個host:port組成
• acks：broker訊息確認的模式，有三種：
  0：不進行訊息接收確認，即Client端傳送完成後不會等待Broker的確認
  1：由Leader確認，Leader接收到訊息後會立即返回確認資訊
  all：叢集完整確認，Leader會等待所有in-sync的follower節點都確認收到訊息後，再返回確認資訊
  我們可以根據訊息的重要程度，設定不同的確認模式。預設為1
• retries：傳送失敗時Producer端的重試次數，預設為0
• batch.size：當同時有大量訊息要向同一個分割槽傳送時，Producer端會將訊息打包後進行批量傳送。如果設定為0，則每條訊息都獨立傳送。預設為16384位元組
• linger.ms：傳送訊息前等待的毫秒數，與batch.size配合使用。在訊息負載不高的情況下，配置linger.ms能夠讓Producer在傳送訊息前等待一定時間，以積累更多的訊息打包傳送，達到節省網路資源的目的。預設為0
• key.serializer/value.serializer：訊息key/value的序列器Class，根據key和value的型別決定
• buffer.memory：訊息緩衝池大小。尚未被髮送的訊息會儲存在Producer的記憶體中，如果訊息產生的速度大於訊息傳送的速度，那麼緩衝池滿後傳送訊息的請求會被阻塞。預設33554432位元組（32MB）

  更多的Producer配置見官網：http://kafka.apache.org/documentation.html#producerconfigs

  3) 寫一個簡單的Producer端，每隔1秒向Kafka叢集傳送一條訊息：

Java程式碼 

1. publicclass KafkaTest {  
2.     publicstaticvoid main(String[] args) throws Exception{  
3.         Producer<String, String> producer = KafkaUtil.getProducer();  
4.         int i = 0;  
5.         while(true) {  
6.             ProducerRecord<String, String> record = new ProducerRecord<String, String>("test", String.valueOf(i), "this is message"+i);  
7.             producer.send(record, new Callback() {  
8.                 publicvoid onCompletion(RecordMetadata metadata, Exception e) {  
9.                     if (e != null)  
10.                         e.printStackTrace();  
11.                     System.out.println("message send to partition " + metadata.partition() + ", offset: " + metadata.offset());  
12.                 }  
13.             });  
14.             i++;  
15.             Thread.sleep(1000);  
16.         }  
17.     }  
18. }  

  在呼叫KafkaProducer的send方法時，可以註冊一個回撥方法，在Producer端完成傳送後會觸發回撥邏輯，在回撥方法的 metadata物件中，我們能夠獲取到已傳送訊息的offset和落在的分割槽等資訊。注意，如果acks配置為0，依然會觸發回撥邏輯，只是拿不到 offset和訊息落地的分割槽資訊。

    跑一下，輸出是這樣的：

message send to partition 0, offset: 28
message send to partition 1, offset: 26
message send to partition 0, offset: 29
message send to partition 1, offset: 27
message send to partition 1, offset: 28
message send to partition 0, offset: 30
message send to partition 0, offset: 31
message send to partition 1, offset: 29
message send to partition 1, offset: 30
message send to partition 1, offset: 31
message send to partition 0, offset: 32
message send to partition 0, offset: 33
message send to partition 0, offset: 34
message send to partition 1, offset: 32

  乍一看似乎offset亂掉了，但其實這是因為訊息分佈在了兩個分割槽上，每個分割槽上的offset其實是正確遞增的。

5. 使用Kafka的Consumer API來完成訊息的消費

1) 改造一下KafkaUtil類，加入Consumer client的構造。

Java程式碼 

1. publicclass KafkaUtil {  
2.     privatestatic KafkaProducer<String, String> kp;  
3.     privatestatic KafkaConsumer<String, String> kc;  
4.     publicstatic KafkaProducer<String, String> getProducer() {  
5.         if (kp == null) {  
6.             Properties props = new Properties();  
7.             props.put("bootstrap.servers", "10.0.0.100:9092,10.0.0.101:9092");  
8.             props.put("acks", "1");  
9.             props.put("retries", 0);  
10.             props.put("batch.size", 16384);  
11.             props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
12.             props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
13.             kp = new KafkaProducer<String, String>(props);  
14.         }  
15.         return kp;  
16.     }  
17.     publicstatic KafkaConsumer<String, String> getConsumer() {  
18.         if(kc == null) {  
19.             Properties props = new Properties();  
20.             props.put("bootstrap.servers", "10.0.0.100:9092,10.0.0.101:9092");  
21.             props.put("group.id", "1");  
22.             props.put("enable.auto.commit", "true");  
23.             props.put("auto.commit.interval.ms", "1000");  
24.             props.put("session.timeout.ms", "30000");  
25.             props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
26.             props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
27.             kc = new KafkaConsumer<String, String>(props);  
28.         }  
29.         return kc;  
30.     }  
31. }  

  同樣，我們介紹一下Consumer常用配置

• bootstrap.servers/key.deserializer/value.deserializer：和Producer端的含義一樣，不再贅述
• fetch.min.bytes：每次最小拉取的訊息大小（byte）。Consumer會等待訊息積累到一定尺寸後進行批量拉取。預設為1，代表有一條就拉一條
• max.partition.fetch.bytes：每次從單個分割槽中拉取的訊息最大尺寸（byte），預設為1M
• group.id：Consumer的group id，同一個group下的多個Consumer不會拉取到重複的訊息，不同group下的Consumer則會保證拉取到每一條訊息。注意，同一個group下的consumer數量不能超過分割槽數。
• enable.auto.commit：是否自動提交已拉取訊息的offset。提交offset即視為該訊息已經成功被消費，該組下的Consumer無法再拉取到該訊息（除非手動修改offset）。預設為true
• auto.commit.interval.ms：自動提交offset的間隔毫秒數，預設5000。

  全部的Consumer配置見官方文件：http://kafka.apache.org/documentation.html#newconsumerconfigs

2) 編寫Consumer端：

Java程式碼 

1. publicclass KafkaTest {  
2.     publicstaticvoid main(String[] args) throws Exception{  
3.         KafkaConsumer<String, String> consumer = KafkaUtil.getConsumer();  
4.         consumer.subscribe(Arrays.asList("test"));  
5.         while(true) {  
6.             ConsumerRecords<String, String> records = consumer.poll(1000);  
7.             for(ConsumerRecord<String, String> record : records) {  
8.                 System.out.println("fetched from partition " + record.partition() + ", offset: " + record.offset() + ", message: " + record.value());  
9.             }  
10.         }  
11.     }  
12. }  

  執行輸出：

fetched from partition 0, offset: 28, message: this is message0
fetched from partition 0, offset: 29, message: this is message2
fetched from partition 0, offset: 30, message: this is message5
fetched from partition 0, offset: 31, message: this is message6
fetched from partition 0, offset: 32, message: this is message10
fetched from partition 0, offset: 33, message: this is message11
fetched from partition 0, offset: 34, message: this is message12
fetched from partition 1, offset: 26, message: this is message1
fetched from partition 1, offset: 27, message: this is message3
fetched from partition 1, offset: 28, message: this is message4
fetched from partition 1, offset: 29, message: this is message7
fetched from partition 1, offset: 30, message: this is message8
fetched from partition 1, offset: 31, message: this is message9
fetched from partition 1, offset: 32, message: this is message13

說明：

• KafkaConsumer的poll方法即是從Broker拉取訊息，在poll之前首先要用subscribe方法訂閱一個Topic。
• poll方法的入參是拉取超時毫秒數，如果沒有新的訊息可供拉取，consumer會等待指定的毫秒數，到達超時時間後會直接返回一個空的結果集。
• 如 果Topic有多個partition，KafkaConsumer會在多個partition間以輪詢方式實現負載均衡。如果啟動了多個 Consumer執行緒，Kafka也能夠通過zookeeper實現多個Consumer間的排程，保證同一組下的Consumer不會重複消費訊息。注 意，Consumer數量不能超過partition數，超出部分的Consumer無法拉取到任何資料。
• 可以看出，拉取到的訊息並不是完全順序化的，kafka只能保證一個partition內的訊息先進先出，所以在跨partition的情況下，訊息的順序是沒有保證的。
• 本 例中採用的是自動提交offset，Kafka client會啟動一個執行緒定期將offset提交至broker。假設在自動提交的間隔內發生故障（比如整個JVM程序死掉），那麼有一部分訊息是會被 重複消費的。要避免這一問題，可使用手動提交offset的方式。構造consumer時將enable.auto.commit設為false，並在代 碼中用consumer.commitSync()來手動提交。

如果不想讓kafka控制consumer拉取資料時在partition間的負載均衡，也可以手工控制：

Java程式碼 

1. publicstaticvoid main(String[] args) throws Exception{  
2.     KafkaConsumer<String, String> consumer = KafkaUtil.getConsumer();  
3.     String topic = "test";  
4.     TopicPartition partition0 = new TopicPartition(topic, 0);  
5.     TopicPartition partition1 = new TopicPartition(topic, 1);  
6.     consumer.assign(Arrays.asList(partition0, partition1));  
7.     while(true) {  
8.         ConsumerRecords<String, String> records = consumer.poll(100);  
9.         for(ConsumerRecord<String, String> record : records) {  
10.             System.out.println("fetched from partition " + record.partition() + ", offset: " + record.offset() + ", message: " + record.value());  
11.         }  
12.         consumer.commitSync();  
13.     }  
14. }  

 使用consumer.assign()方法為consumer執行緒指定1個或多個partition。

  此處的坑：

在測試中我發現，如果用手工指定partition的方法拉取訊息，不知為何kafka的自動提交offset機制會失效，必須使用手動方式才能正確提交已消費的訊息offset。

  題外話：

在 真正的應用環境中，Consumer端將訊息拉取下來後要做的肯定不止是輸出出來這麼簡單，在消費訊息時很有可能需要花掉更多的時間。1個 Consumer執行緒消費訊息的速度很有可能是趕不上Producer產生訊息的速度，所以我們不得不考慮Consumer端採用多執行緒模型來消費訊息。
然而KafkaConsumer並不是執行緒安全的，多個執行緒操作同一個KafkaConsumer例項會出現各種問題，Kafka官方對於Consumer端的多執行緒處理給出的指導建議如下：

1. 每個執行緒都持有一個KafkaConsumer物件
好處：

• 實現簡單
• 不需要執行緒間的協作，效率最高
• 最容易實現每個Partition內訊息的順序處理

弊端：

• 每個KafkaConsumer都要與叢集保持一個TCP連線
• 執行緒數不能超過Partition數
• 每一batch拉取的資料量會變小，對吞吐量有一定影響

2. 解耦，1個Consumer執行緒負責拉取訊息，數個Worker執行緒負責消費訊息
好處：

• 可自由控制Worker執行緒的數量，不受Partition數量限制

弊端：

• 訊息消費的順序無法保證
• 難以控制手動提交offset的時機

個人認為第二種方式更加可取，consumer數不能超過partition數這個限制是很要命的，不可能為了提高Consumer消費訊息的效率而把Topic分成更多的partition，partition越多，叢集的高可用性就越低。

6. Kafka叢集高可用性測試

1) 檢視當前Topic的狀態：

Shell程式碼 

1. /kafka-topics.sh --describe --zookeeper 10.0.0.100:2181,10.0.0.101:2181,10.0.0.102:2181 --topic test  

  輸出：

Topic:test PartitionCount:2 ReplicationFactor:2 Configs:
   Topic: test Partition: 0 Leader: 1 Replicas: 1,0 Isr: 0,1
   Topic: test Partition: 1 Leader: 0 Replicas: 0,1 Isr: 0,1

  可以看到，partition0的leader是broker1，parition1的leader是broker0

2) 啟動Producer向Kafka叢集傳送訊息

  輸出：

message send to partition 0, offset: 35
message send to partition 1, offset: 33
message send to partition 0, offset: 36
message send to partition 1, offset: 34
message send to partition 1, offset: 35
message send to partition 0, offset: 37
message send to partition 0, offset: 38
message send to partition 1, offset: 36
message send to partition 1, offset: 37

3) 登入SSH將broker0，也就是partition 1的leader kill掉

  再次檢視Topic狀態：

Topic:test PartitionCount:2 ReplicationFactor:2 Configs:
  Topic: test Partition: 0 Leader: 1 Replicas: 1,0 Isr: 1
  Topic: test Partition: 1 Leader: 1 Replicas: 0,1 Isr: 1

  可以看到，當前parition0和parition1的leader都是broker1了

  此時再去看Producer的輸出：

[kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.common.network.Selector - Connection with /10.0.0.100 disconnected
java.net.ConnectException: Connection refused: no further information
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:739)
    at org.apache.kafka.common.network.PlaintextTransportLayer.finishConnect(PlaintextTransportLayer.java:54)
    at org.apache.kafka.common.network.KafkaChannel.finishConnect(KafkaChannel.java:72)
    at org.apache.kafka.common.network.Selector.poll(Selector.java:274)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:256)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:128)
    at java.lang.Thread.run(Thread.java:745)
[kafka-producer-network-thread | producer-1] DEBUG org.apache.kafka.clients.Metadata - Updated cluster metadata version 7 to Cluster(nodes = [Node(1, 10.0.0.101, 9092)], partitions = [Partition(topic = test, partition = 1, leader = 1, replicas = [1,], isr = [1,], Partition(topic = test, partition = 0, leader = 1, replicas = [1,], isr = [1,]])

  能看到Producer端的DEBUG日誌顯示與broker0的連結斷開了，此時Kafka立刻開始更新叢集metadata，更新後的metadata表示broker1現在是兩個partition的leader，Producer程序很快就恢復繼續執行，沒有漏發任何訊息，能夠看出Kafka叢集的故障切換機制還是很厲害的

4) 我們再把broker0啟動起來

Shell程式碼 

1. bin/kafka-server-start.sh -daemon config/server.properties   

  然後再次檢查Topic狀態：

Topic:test PartitionCount:2 ReplicationFactor:2 Configs:
   Topic: test Partition: 0 Leader: 1 Replicas: 1,0 Isr: 1,0
   Topic: test Partition: 1 Leader: 1 Replicas: 0,1 Isr: 1,0

  我們看到，broker0啟動起來了，並且已經是in-sync狀態（注意Isr從1變成了1,0），但此時兩個partition的leader還都是 broker1，也就是說當前broker1會承載所有的傳送和拉取請求。這顯然是不行的，我們要讓叢集恢復到負載均衡的狀態。

  這時候，需要使用Kafka的選舉工具觸發一次選舉：

Shell程式碼 

1. bin/kafka-preferred-replica-election.sh --zookeeper 10.0.0.100:2181,10.0.0.101:2181,10.0.0.102:2181

  選舉完成後，再次檢視Topic狀態：

Topic:test PartitionCount:2 ReplicationFactor:2 Configs:
   Topic: test Partition: 0 Leader: 1 Replicas: 1,0 Isr: 1,0
   Topic: test Partition: 1 Leader: 0 Replicas: 0,1 Isr: 1,0

  可以看到，叢集重新回到了broker0掛掉之前的狀態

  但此時，Producer端產生了異常：

org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition.

  原因是Producer端在嘗試向broker1的parition0傳送訊息時，partition0的leader已經切換成了broker0，所以訊息傳送失敗。

  此時用Consumer去消費訊息，會發現訊息的編號不連續了，確實漏發了一條訊息。這是因為我們在構造Producer時設定了retries=0，所以在傳送失敗時Producer端不會嘗試重發。

  將retries改為3後再次嘗試，會發現leader切換時再次發生了同樣的問題，但Producer的重發機制起了作用，訊息重發成功，啟動Consumer端檢查也證實了所有訊息都發送成功了。

每 次叢集單點發生故障恢復後，都需要進行重新選舉才能徹底恢復叢集的leader分配，如果嫌每次這樣做很麻煩，可以在broker的配置檔案（即 server.properties）中配置auto.leader.rebalance.enable=true，這樣broker在啟動後就會自動進 行重新選舉

至此，我們通過測試證實了叢集出現單點故障和恢復的過程中，Producer端能夠保持正確運轉。接下來我們看一下Consumer端的表現：

5) 同時啟動Producer程序和Consumer程序

  此時Producer一邊在生產訊息，Consumer一邊在消費訊息

6) 把broker0幹掉，觀察Consumer端的輸出：

能看到，在broker0掛掉後，consumer也端產生了一系列INFO和WARN輸出，但同Producer端一樣，若干秒後自動恢復，訊息仍然是連續的，並未出現斷點。

7) 再次把broker0啟動，並觸發重新選舉，然後觀察輸出：

fetched from partition 0, offset: 418, message: this is message48
fetched from partition 0, offset: 419, message: this is message49
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - Offset commit for group 1 failed due to NOT_COORDINATOR_FOR_GROUP, will find new coordinator and retry
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - Marking the coordinator 2147483646 dead.
[main] WARN org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - Auto offset commit failed: This is not the correct coordinator for this group.
fetched from partition 1, offset: 392, message: this is message50
fetched from partition 0, offset: 420, message: this is message51

  能看到，重選舉後Consumer端也輸出了一些日誌，意思是在提交offset時發現當前的排程器已經失效了，但很快就重新獲取了新的有效排程器，恢復 了offset的自動提交，驗證已提交offset的值也證明了offset提交併未因leader切換而發生錯誤。

  如上，我們也通過測試證實了Kafka叢集出現單點故障時，Consumer端的功能正確性。