kafka7 探索生產者同步or異步發送消息

本文轉載自查看原文 2018-12-21 16:17 7268 bigdata

1.生產者：在發送完消息后，收到回執確認。

主要是在SimpleProducer.java中修改了發送消息的2行代碼，用到了回調函數，修改如下：

//發送消息
ProducerRecord<String, String> rec = new ProducerRecord<String, String>("test-topic","hello world from win7");
producer.send(rec,new Callback() {
    public void onCompletion(RecordMetadata metadata,Exception exception) {
        System.out.println("ack!!!");
    }
}); //在發送消息后，收到回執確認。

完整代碼如下：

 1 package cn.test.mykafka;
 2 
 3 import java.util.Properties;
 4 
 5 import org.apache.kafka.clients.producer.Callback;
 6 import org.apache.kafka.clients.producer.KafkaProducer;
 7 import org.apache.kafka.clients.producer.Producer;
 8 import org.apache.kafka.clients.producer.ProducerRecord;
 9 import org.apache.kafka.clients.producer.RecordMetadata;
10 
11 /**
12  * 簡單生產者:在發送完消息后，收到回執確認。
13  *
14  */
15 
16 public class SimpleProducer2 {
17 
18     public static void main(String[] args) {
19         
20          //創建配置信息
21          Properties props = new Properties();
22          props.put("bootstrap.servers", "192.168.42.133:9092"); //指定broker的節點和端口
23          props.put("acks", "all");
24          props.put("retries", 0);
25          props.put("batch.size", 16384);
26          props.put("linger.ms", 1);
27          props.put("buffer.memory", 33554432);
28          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
29          props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
30 
31          //創建一個生產者
32          Producer<String, String> producer = new KafkaProducer<>(props);
33          
34         //發送消息
35          ProducerRecord<String, String> rec = new ProducerRecord<String, String>("test-topic","hello world from win7");
36          producer.send(rec,new Callback() {
37              public void onCompletion(RecordMetadata metadata,Exception exception) {
38                  System.out.println("ack!!!");
39              }
40          }); //在發送消息后，收到回執確認。
41          
42          //for (int i = 0; i < 10; i++)
43          //   producer.send(new ProducerRecord<String, String>("test-topic", Integer.toString(i), Integer.toString(i))); //topic,key(非必填),value 
44             
45          System.out.println("over");
46          producer.close();
47     }
48 }

SimpleProducer2.java

2.比較同步和異步生產者消息發送速度。

完整代碼如下：

 1 package cn.test.mykafka;
 2 
 3 import java.util.Properties;
 4 
 5 import org.apache.kafka.clients.producer.Callback;
 6 import org.apache.kafka.clients.producer.KafkaProducer;
 7 import org.apache.kafka.clients.producer.Producer;
 8 import org.apache.kafka.clients.producer.ProducerRecord;
 9 import org.apache.kafka.clients.producer.RecordMetadata;
10 
11 /**
12  * 比較同步和異步生產者消息發送速度，參數未生效，失敗
13  *
14  */
15 
16 public class SimpleProducer3 {
17 
18     static long starttime;
19     public static void main(String[] args) {
20         
21          //創建配置信息
22          Properties props = new Properties();
23          props.put("bootstrap.servers", "192.168.42.133:9092"); //指定broker的節點和端口
24          props.put("acks", "all");
25          props.put("retries", 0);
26          props.put("batch.size", 16384);
27          props.put("linger.ms", 1);
28          props.put("buffer.memory", 33554432);
29          props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
30          props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
31          props.put("producer.type", "async");  //sync為同步，async為異步,此配置未生效
32          
33          //創建一個生產者
34          Producer<String, String> producer = new KafkaProducer<>(props);
35          
36          StringBuilder builder = new StringBuilder();
37          for( int i = 0 ; i < 100000 ; i++) {
38              builder.append(" " + i + ",");
39          }
40          
41         //發送消息
42          ProducerRecord<String, String> rec = new ProducerRecord<String, String>("test-topic",builder.toString());
43          producer.send(rec,new Callback() {
44              public void onCompletion(RecordMetadata metadata,Exception exception) {
45                  System.out.println("receive ack : "+ (System.currentTimeMillis()-starttime) + "ms");
46              }
47          }); //在發送消息后，收到回執確認
48          
49          starttime = System.currentTimeMillis();
50          System.out.println("over");
51          producer.close();
52     }
53 }

SimpleProducer3.java

kafka同步生產者：這個生產者寫一條消息的時候，它就立馬發送到某個分區去。follower還需要從leader拉取消息到本地，follower再向leader發送確認，leader再向客戶端發送確認。由於這一套流程之后，客戶端才能得到確認，所以很慢。
kafka異步生產者：這個生產者寫一條消息的時候，先是寫到某個緩沖區，這個緩沖區里的數據還沒寫到broker集群里的某個分區的時候，它就返回到client去了。雖然效率快，但是不能保證消息一定被發送出去了。

客戶端向topic發送數據分為兩種方式：
producer.type=sync 同步模式
producer.type=async 異步模式

執行以上代碼（通過控制producer.type參數取值sync/async）時，輸出警告：

WARN org.apache.kafka.clients.producer.ProducerConfig - The configuration 'producer.type' was supplied but isn't a known config.

這說明producer.type參數配置根本沒生效，后來我在官方文檔中也沒有找到這個參數，估計在kafka 2.0.0版本中此參數已經沒有了。

於是乎，我在網上找了另一段代碼（參考博客），修改后如下：

  1 package cn.test.mykafka;
  2 
  3 import java.util.Properties;
  4 import java.util.concurrent.Future;
  5 
  6 import org.apache.kafka.clients.producer.Callback;
  7 import org.apache.kafka.clients.producer.KafkaProducer;
  8 import org.apache.kafka.clients.producer.Producer;
  9 import org.apache.kafka.clients.producer.ProducerRecord;
 10 import org.apache.kafka.clients.producer.RecordMetadata;
 11 
 12 /**
 13  * 比較同步和異步生產者消息發送速度，有2個問題: 1.我不缺定starttime的取值位置是否正確？ 2.時間差不多，沒得出啥結論
 14  *
 15  */
 16 
 17 public class SimpleProducer4 {
 18 
 19     static long starttime;
 20 
 21     StringBuilder builder = new StringBuilder();
 22 
 23     public void initBuilder() {
 24         for (int i = 0; i < 100000; i++) {
 25             builder.append(" " + i + ",");
 26         }
 27     }
 28 
 29     private Properties kafkaProps = new Properties();
 30 
 31     /**
 32      * 初始化一些配置信息
 33      */
 34     public void initProperty() {
 35         kafkaProps.put("bootstrap.servers", "192.168.42.133:9092"); // 指定broker的節點和端口
 36         kafkaProps.put("acks", "all");
 37         kafkaProps.put("retries", 0);
 38         kafkaProps.put("batch.size", 16384);
 39         kafkaProps.put("linger.ms", 1);
 40         kafkaProps.put("buffer.memory", 33554432);
 41         kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 42         kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 43     }
 44 
 45     /**
 46      * 加載配置信息，生成一個生產者實例
 47      * 
 48      * @param props
 49      * @return
 50      */
 51     public Producer<String, String> getProducer(Properties props) {
 52         if (props == null || props.size() == 0)
 53             throw new IllegalArgumentException();
 54         return new KafkaProducer<>(props);
 55     }
 56 
 57     /**
 58      * 同步發送消息
 59      * 
 60      * @param producer
 61      * @throws Exception
 62      */
 63     public void syncSend(Producer<String, String> producer) throws Exception {
 64 
 65         ProducerRecord<String, String> record = new ProducerRecord<String, String>("test-topic",
 66                 builder.toString() + "this is a sygn record");
 67 
 68         // 同步發送消息，消息發送成功后，服務端會返回給一個RecordMetadata對象
 69         Future<RecordMetadata> future = producer.send(record);
 70         starttime = System.currentTimeMillis();
 71         RecordMetadata metadata = future.get();
 72 
 73         System.out.println("offset:" + metadata.offset() + "\npartition:" + metadata.partition() + "\ntopic:"
 74                 + metadata.topic() + "\nserializedKeySize:" + metadata.serializedKeySize() + "\nserializedValueSize:"
 75                 + metadata.serializedValueSize() + "\nreceive sygn ack : " + (System.currentTimeMillis() - starttime)
 76                 + "ms" + "\n");
 77 
 78         producer.close();
 79     }
 80 
 81     /**
 82      * 異步發送消息
 83      * 
 84      * @param producer
 85      */
 86     public void asyncSend(Producer<String, String> producer) {
 87 
 88         ProducerRecord<String, String> record = new ProducerRecord<String, String>("test-topic",
 89                 builder.toString() + "this is a asygn record");
 90 
 91         producer.send(record, new Callback() {
 92             public void onCompletion(RecordMetadata metadata, Exception e) {
 93                 System.out.println("offset:" + metadata.offset() + "\npartition:" + metadata.partition() + "\ntopic:"
 94                         + metadata.topic() + "\nserializedKeySize:" + metadata.serializedKeySize()
 95                         + "\nserializedValueSize:" + metadata.serializedValueSize() + "\n");
 96                 if (e == null) {
 97                     System.out.println("\nreceive asygn ack : " + (System.currentTimeMillis() - starttime) + "ms");
 98                 }
 99             }
100         });
101         starttime = System.currentTimeMillis();
102         producer.close();
103     }
104 
105     public void start() throws Exception {
106         initBuilder();
107         initProperty();
108         // syncSend(getProducer(kafkaProps));
109         asyncSend(getProducer(kafkaProps));
110 
111     }
112 
113     public static void main(String[] args) throws Exception {
114 
115         SimpleProducer4 myProducer = new SimpleProducer4();
116         myProducer.start();
117     }
118 }

SimpleProducer4.java

執行之后，發現同步和異執行時間差不多，沒能證明異步比同步快。推測原因有2個：

1.消息太短，存在偶然性，看不出時間差；

2.我不確定這段代碼是否正確？同步代碼是否正確？異步代碼是否正確？開始時間取值位置是否正確？

好吧，心好累。關於生產者的同步異步問題就先到這里吧，雖然結果不盡人意，但是過程中也學到了很多，以后有時間再繼續吧。

PS：以上純粹是我的探索測試，如果有不對的地方，歡迎留言指正，不勝感激。

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 kafka同步生產者和異步生產者深入剖析 Kafka生產者發送消息的三種方式 Kafka sender消息生產者 kafka（java客戶端）消費者取不到消息，生產者消息也沒發送成功【轉】kafka（java客戶端）消費者取不到消息，生產者消息也沒發送成功 kafka 同步發送消息【kafka】生產者API 回調同步 kafka 0.8.2 消息生產者 producer kafka 0.8.2 消息生產者 KafkaProducer Kafka生產者——發送原理分析