关于kafka发送消息的三种方式总结

package com.zl.kafkademo; 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 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(properties); } /** * 第一种直接发送，不管结果 */ private static void sendMessageForgetResult(){ ProducerRecord record = new ProducerRecord( "kafka-study","name","Forget_result" ); producer.send(record); producer.close(); } /** * 第二种同步发送，等待执行结果 * @return * @throws Exception */ private static RecordMetadata sendMessageSync() throws Exception{ ProducerRecord record = new ProducerRecord( "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 record = new ProducerRecord( "kafka-study","name","callback" ); producer.send(record,new MyProducerCallback()); } //定时任务 @Override public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException { try { sendMessageSync(); }catch (Exception e){ System.out.println("error:"+e); } } private static class MyProducerCallback implements Callback{ @Override public void onCompletion(RecordMetadata recordMetadata, Exception e) { if (e !=null){ e.printStackTrace(); return; } System.out.println(recordMetadata.topic()); System.out.println(recordMetadata.partition()); System.out.println(recordMetadata.offset()); System.out.println("Coming in MyProducerCallback"); } } public static void main(String[] args){ //sendMessageForgetResult(); //sendMessageCallback(); JobDetail job = JobBuilder.newJob(MyProducer.class).build(); Trigger trigger = TriggerBuilder.newTrigger() .withSchedule(SimpleScheduleBuilder.repeatSecondlyForever()).build(); try { Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler(); scheduler.scheduleJob(job,trigger); scheduler.start(); }catch (SchedulerException e){ e.printStackTrace(); } catch (Exception e) { e.printStackTrace(); } } }

         org.apache.kafkakafka-clients0.10.0.1 org.quartz-schedulerquartz2.3.0

./kafka-topics.sh --create --topic kafka-study --zookeeper 127.0.0.1:2181 --config max.message.bytes=12800000 --config flush.messages=1 --partitions 5 --replication-factor 1

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kafka-study --from-beginning

.\bin\windows\kafka-server-start.bat .\config\server.properties

.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

.\bin\windows\kafka-topics.bat --list --zookeeper localhost:2181

.\bin\windows\kafka-topics.bat --describe --zookeeper localhost:2181 --topic test

.\bin\windows\kafka-console-consumer.bat --zookeeper localhost:2181 --topic kafka-study --from-beginning