生產(chǎn)者發(fā)送消息調(diào)用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);}/*** 第一種直接發(fā)送，不管結(jié)果 有些異常捕捉不到 異步發(fā)送*/private static void sendMessageForgetResult(){ProducerRecord<String,String> record = new ProducerRecord<String,String>("kafka-study","name","Forget_result");producer.send(record);producer.close();}/*** 第二種同步發(fā)送，等待執(zhí)行結(jié)果 同步發(fā)送* @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;}/*** 第三種執(zhí)行回調(diào)函數(shù) 異步發(fā)送*/private static void sendMessageCallback(){ProducerRecord<String,String> record = new ProducerRecord<String,String>("kafka-study","name","callback");producer.send(record,new MyProducerCallback());}

?

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。