kafka7 探索生产者同步or异步发送消息
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:以上纯粹是我的探索测试,如果有不对的地方,欢迎留言指正,不胜感激。
转载于:https://www.cnblogs.com/zhengna/p/9948709.html
kafka7 探索生产者同步or异步发送消息相关推荐
- kafka实现异步发送_Kafka Producer 异步发送消息居然也会阻塞?
Kafka 一直以来都以高吞吐量的特性而家喻户晓,就在上周,在一个性能监控项目中,需要使用到 Kafka 传输海量消息,在这过程中遇到了一个 Kafka Producer 异步发送消息会被阻塞的问题, ...
- Kafka生产者——消息发送流程,同步、异步发送API
生产者消息发送流程 发送原理 Kafka的Producer发送消息采用的是异步发送的方式. 在消息发送的过程中,涉及到了两个线程:main线程和Sender线程,以及一个线程共享变量:RecordAc ...
- 驱动中同步与异步发送IRP
1. 同步 同步创建方式的意思是应用层必须要等待IRP被IoCompleteRequest后才会返回.其必须是同步的.即使底层设备对收到的IRP进行挂起处理,那也得无限等待下去.看一下下面的例子: 该 ...
- 消息的同步发送,异步发送以及消息发送的可靠性
最近写的一个通信框架中有两种最基本的消息发送方式:同步发送和异步发送. 同步方式: 消息的发送方发A送一条消息到接收端B,B收到消息之后需要对消息进行处理,然后发送ACK确认消息回A,A收到B的ACK ...
- 【Kafka】消息的同步发送和异步发送
文章目录 概述 1. sync vs async 1.1 java代码同步和异步 2. 可靠性机制(ack属性配置) 2.1 oneway 3. 一般配置 4. 同步异步和ack的联系和区别 参考 概 ...
- 异步消息服务器,Spring-Kafka中如何通过KafkaTemplate 对象异步和同步发送消息
Kafka Producer默认是异步发送. 在初始化producer实例时,会创建一个sender线程负责批量发送消息: producer将消息暂存在缓冲区,消息根据topic-partition分 ...
- kafka实现异步发送_Kafka 异步消息也会阻塞?记一次 Dubbo 频繁超时排查过程
线上某服务 A 调用服务 B 接口完成一次交易,一次晚上的生产变更之后,系统监控发现服务 B 接口频繁超时,后续甚至返回线程池耗尽错误 Thread pool is EXHAUSTED.因为服务 B ...
- Kafka生产者发送消息的三种方式
Kafka是一种分布式的基于发布/订阅的消息系统,它的高吞吐量.灵活的offset是其它消息系统所没有的. Kafka发送消息主要有三种方式: 1.发送并忘记 2.同步发送 3.异步发送+回调函数 下 ...
- 【kafka系列】kafka之生产者发送消息实践
目录 一.准备工作 二.终端命令 生产者命令 消费者命令 三.Java实践 搭建项目 异步发送-无回调 异步发送-有回调 同步发送 一.准备工作 进入实战之前先熟悉一下topic的相关命令,使用终端命 ...
- Kafka生产者是如何发送消息的?
(一)生产者的原理 当有数据要从生产者发往消费者的时候,在kafka底层有这样一套流程.首先生产者调用send方法发送消息后,会先经过一层拦截器,接着进入序列化器.序列化器主要用于对消息的Key和Va ...
最新文章
- (前)首富许家印造车果然有一套:车还没量产上市,公司先冲刺科创板募资
- 各种initcall的执行先后顺序(module_init、postcore_initcall、arch_initcall、subsys_initcall、 fs_initcall)...
- soj1209- 最短的距离(精度问题)
- 天翼云从业认证(1.1)服务器的分类、用途、特点、结构和组件
- 两台linux之间传递文件
- noip2017d2t2
- Oracle入门(十四F)之PL/SQL定义变量
- Mr.J--Java基础问题30问
- Java关键字synchronized的简单理解
- (32)Verilog HDL按位运算
- fast.ai 深度学习笔记:第一部分第二课
- Linux基础第六章 信号
- 天思经理人ERP日化行业应用方案
- Linux ntp时间服务器的搭建和配置
- JAVA8的一些写法
- Win10无法开机提示自动修复无法修复你的电脑的有效解决方法
- (19年最新,操作极简)linux下使用xmind zen破解版
- 观影感受 之 《绿皮书》
- 春天喝这五种花草茶 身体排毒又美容养颜
- uri (url)保存为jpg图片(文件)