生產者發送消息調用send方法

下面介紹了官方提供的三種方式。

import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.quartz.*; import org.quartz.impl.StdSchedulerFactory;import java.util.Properties;/*** @Auther: le* @Date: 2019/4/23 22:05* @Description:*/ public class MyProducer implements Job {private static KafkaProducer<String,String> producer;static {Properties properties = new Properties();properties.put("bootstrap.servers","127.0.0.1:9092");properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");producer = new KafkaProducer<String, String>(properties);}/*** 第一種直接發送，不管結果 有些異常捕捉不到 異步發送*/private static void sendMessageForgetResult(){ProducerRecord<String,String> record = new ProducerRecord<String,String>("kafka-study","name","Forget_result");producer.send(record);producer.close();}/*** 第二種同步發送，等待執行結果 同步發送* @return* @throws Exception*/private static RecordMetadata sendMessageSync() throws Exception{ProducerRecord<String,String> record = new ProducerRecord<String,String>("kafka-study","name","sync");RecordMetadata result = producer.send(record).get();System.out.println(result.topic());System.out.println(result.partition());System.out.println(result.offset());return result;}/*** 第三種執行回調函數 異步發送*/private static void sendMessageCallback(){ProducerRecord<String,String> record = new ProducerRecord<String,String>("kafka-study","name","callback");producer.send(record,new MyProducerCallback());}

?

總結

以上是生活随笔為你收集整理的kafka send方法详解 （同步异步） 小白使用记录的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。