kafka之java客户端实战

1. kafka的客户端

        Kafka提供了两套客户端API，HighLevel API和LowLevel API。 HighLevel API封装了kafka的运行细节，使用起来比较简单，是企业开发过程中最常用的客户端API。 而LowLevel API则需要客户端自己管理Kafka的运行细节，Partition，Offset这些数据都由客户端自行管理。这层API功能更灵活，但是使用起来非常复杂，也更容易出错。只在极少数对性能要求非常极致的场景才会偶尔使用。我们的重点是HighLeve API 。

2. 基础客户端的使用

Kafka提供了非常简单的客户端API。只需要引入一个Maven依赖即可：

  <dependency><groupId>org.apache.kafka</groupId><artifactId>kafka_2.13</artifactId><version>3.4.0</version></dependency>

2.1 如何发消息

        现在， 我们使用Kafka提供的Producer类，如何发送消息。

2.1.1 单项发送消息

代码：

public class MyProducerTest {private static final String BOOTSTRAP_SERVERS = "192.168.31.5:9092,192.168.31.176:9092,192.168.31.232:9092";private static final String TOPIC = "disTopic";public static void main(String[] args) throws ExecutionException, InterruptedException {//PART1:设置发送者相关属性Properties props = new Properties();// 此处配置的是kafka的端口props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.roy.kfk.basic.MyInterceptor");// 配置key的序列化类props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");// 配置value的序列化类props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");Producer<String,String> producer = new KafkaProducer<>(props);for(int i = 0; i < 5; i++) {//Part2:构建消息ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, Integer.toString(i), "MyProducer" + i);//Part3:发送消息//单向发送：不关心服务端的应答。producer.send(record);System.out.println("message "+i+" sended");}//消息处理完才停止发送者。producer.close();}
}

执行结果：

2.1.2 同步发送

代码：

public class MyProducerTest {private static final String BOOTSTRAP_SERVERS = "192.168.31.5:9092,192.168.31.176:9092,192.168.31.232:9092";private static final String TOPIC = "disTopic";public static void main(String[] args) throws ExecutionException, InterruptedException {//PART1:设置发送者相关属性Properties props = new Properties();// 此处配置的是kafka的端口props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.roy.kfk.basic.MyInterceptor");// 配置key的序列化类props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");// 配置value的序列化类props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");Producer<String,String> producer = new KafkaProducer<>(props);for(int i = 0; i < 5; i++) {//Part2:构建消息ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, Integer.toString(i), "MyProducer" + i);//Part3:发送消息//同步发送：获取服务端应答消息前，会阻塞当前线程。RecordMetadata recordMetadata = producer.send(record).get();String topic = recordMetadata.topic();int partition = recordMetadata.partition();long offset = recordMetadata.offset();String message = recordMetadata.toString();System.out.println("message:["+ message+"] sended with topic:"+topic+"; partition:"+partition+ ";offset:"+offset);}//消息处理完才停止发送者。producer.close();}
}

执行结果：

 2.1.2 异步发送 

代码：

public class MyProducerTest {private static final String BOOTSTRAP_SERVERS = "192.168.31.5:9092,192.168.31.176:9092,192.168.31.232:9092";private static final String TOPIC = "disTopic";public static void main(String[] args) throws ExecutionException, InterruptedException {//PART1:设置发送者相关属性Properties props = new Properties();// 此处配置的是kafka的端口props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.roy.kfk.basic.MyInterceptor");// 配置key的序列化类props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");// 配置value的序列化类props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");Producer<String,String> producer = new KafkaProducer<>(props);CountDownLatch latch = new CountDownLatch(5);for(int i = 0; i < 5; i++) {//Part2:构建消息ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, Integer.toString(i), "MyProducer" + i);//Part3:发送消息//异步发送：消息发送后不阻塞，服务端有应答后会触发回调函数producer.send(record, new Callback() {@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if(null != e){System.out.println("消息发送失败,"+e.getMessage());e.printStackTrace();}else{String topic = recordMetadata.topic();long offset = recordMetadata.offset();String message = recordMetadata.toString();System.out.println("message:["+ message+"] sended with topic:"+topic+";offset:"+offset);}latch.countDown();}});}//消息处理完才停止发送者。latch.await();//消息处理完才停止发送者。producer.close();}
}

执行结果：

2.1.3 总结 

​ 从上述示例中，我们可以总结出，构建Producer分为三个步骤：

  1. 设置Producer核心属性 ：Producer可选的属性都可以由ProducerConfig类管理。比如ProducerConfig.BOOTSTRAP_SERVERS_CONFIG属性，显然就是指发送者要将消息发到哪个Kafka集群上。这是每个Producer必选的属性。在ProducerConfig中，对于大部分比较重要的属性，都配置了对应的DOC属性进行描述。
  2. 构建消息：Kafka的消息是一个Key-Value结构的消息。其中，key和value都可以是任意对象类型。其中，key主要是用来进行Partition分区的，业务上更关心的是value。
  3. 使用Producer发送消息：通常用到的就是单向发送、同步发送和异步发送者三种发送方式。

2.2 如何消费消息

        接下来可以使用Kafka提供的Consumer类，快速消费消息。

2.2.1 消费消息

代码：

public class MyConsumerTest {private static final String BOOTSTRAP_SERVERS = "192.168.31.5:9092,192.168.31.176:9092,192.168.31.232:9092";private static final String TOPIC = "disTopic";public static void main(String[] args) {//PART1:设置发送者相关属性Properties props = new Properties();//kafka地址props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);//每个消费者要指定一个groupprops.put(ConsumerConfig.GROUP_ID_CONFIG, "test");//key序列化类props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");//value序列化类props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");Consumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Arrays.asList(TOPIC));while (true) {//PART2:拉取消息// 100毫秒超时时间ConsumerRecords<String, String> records = consumer.poll(Duration.ofNanos(100));//PART3:处理消息for (ConsumerRecord<String, String> record : records) {System.out.println("partition = "+record.partition()+"offset = " + record.offset() + ";key = " + record.key() + "; value= " + record.value());}//提交offset，消息就不会重复推送。consumer.commitSync(); //同步提交，表示必须等到offset提交完毕，再去消费下一批数据。
//            consumer.commitAsync(); //异步提交，表示发送完提交offset请求后，就开始消费下一批数据了。不用等到Broker的确认。}}
}

2.2.2 总结

​ 整体来说，Consumer同样是分为三个步骤：

  1. 设置Consumer核心属性 ：可选的属性都可以由ConsumerConfig类管理。在这个类中，同样对于大部分比较重要的属性，都配置了对应的DOC属性进行描述。同样BOOTSTRAP_SERVERS_CONFIG是必须设置的属性。
  2. 拉取消息：Kafka采用Consumer主动拉取消息的Pull模式。consumer主动从Broker上拉取一批感兴趣的消息。
  3. 处理消息，提交位点：消费者将消息拉取完成后，就可以交由业务自行处理对应的这一批消息了。只是消费者需要向Broker提交偏移量offset。如果不提交Offset，Broker会认为消费者端消息处理失败了，还会重复进行推送。

3. 客户端核心参数与客户端机制

3.1 消费者分组消费机制

3.2 生产者拦截器机制

3.3 消息序列化机制

3.4 消息分区路由机制

3.5 生产者消息缓存机制

3.6 发送应答机制

3.7 生产者消息幂等性

3.8 生产者消息事务

内容更新中

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/409617.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Vim一键配置指南，打造高效率C++开发环境

Vim一键配置指南，打造高效率C++开发环境

文章目录 前言安装与卸载功能演示gcc/g升级问题 前言 Vim作为当下最受欢迎的文本编译器之一&#xff0c;不仅具有强大的文本编辑功能&#xff0c;还提供了高度的可定制性。用户可以根据自己的喜好自定义配置&#xff0c;并且通过自己编写插件或者使用现有的插件来扩展Vim的功能…
阅读更多...
Queue at the School-codeforces

Queue at the School-codeforces

题目链接&#xff1a;Problem - 266B - Codeforces 题目&#xff1a; 解题思路&#xff1a; 大概意思就是一个队伍里有男生女生&#xff0c;男生会不好意思排在女生前而跟后面的女生换位置&#xff0c;一个时间段里换过的男女生就不能再换了&#xff0c;下个时间段再继续判断…
阅读更多...
vite和webpack的区别

vite和webpack的区别

1 构建原理 Webpack 是一个静态模块打包器&#xff0c;通过对项目中的 JavaScript、CSS、图片等文件进行分析&#xff0c;生成对应的静态资源&#xff0c;并且可以通过一些插件和加载器来实现各种功能。Webpack 的主要特点是支持各种复杂的构建场景&#xff0c;例如代码分割、…
阅读更多...
自动化测试框架pytest系列之8个常用的装饰器函数

自动化测试框架pytest系列之8个常用的装饰器函数

自动化测试框架pytest系列之基础概念介绍(一)-CSDN博客 自动化测试框架pytest系列之21个命令行参数介绍(二)-CSDN博客 自动化测试框架pytest系列之强大的fixture功能&#xff0c;为什么fixture强大&#xff1f;一文拆解它的功能参数。(三)-CSDN博客 接上文 3.5 pytest的8…
阅读更多...
Elasticsearch：聊天机器人教程（二）

Elasticsearch：聊天机器人教程（二）

这是继上一篇文章 “Elasticsearch&#xff1a;聊天机器人教程&#xff08;一&#xff09;”的续篇。本教程的这一部分讨论聊天机器人实现中最有趣的方面&#xff0c;以帮助你理解它并对其进行自定义。 数据摄入 在此应用程序中&#xff0c;所有示例文档的摄取都是通过 flask …
阅读更多...
C#判断输入的数字是否符合货币格式

C#判断输入的数字是否符合货币格式

目录 一、用正则表达式判断输入是否符合货币格式 二、用double.TryParse()判断输入是否符合货币格式 一、用正则表达式判断输入是否符合货币格式 // 判断输入是否货币合格 using System.Text.RegularExpressions; namespace IsCurrency_Format {partial class Program{stati…
阅读更多...
Python进程池multiprocessing.Pool

Python进程池multiprocessing.Pool

环境&#xff1a; 鲲鹏920:192核心 内存&#xff1a;756G python&#xff1a;3.9 python单进程的耗时 在做单纯的cpu计算的场景&#xff0c;使用单进程核多进程的耗时做如下测试&#xff1a; 单进程情况下cpu的占用了如下&#xff0c;占用一半的核心数&#xff1a; 每一步…
阅读更多...
【数据结构】归并排序的两种实现方式与计数排序

【数据结构】归并排序的两种实现方式与计数排序

前言&#xff1a;在前面我们讲了各种常见的排序&#xff0c;今天我们就来对排序部分收个尾&#xff0c;再来对归并排序通过递归和非递归的方法进行实现&#xff0c;与对计数排序进行简单的学习。 &#x1f496; 博主CSDN主页:卫卫卫的个人主页 &#x1f49e; &#x1f449; 专栏…
阅读更多...
c#异形窗体遮罩效果

c#异形窗体遮罩效果

c#异形窗体遮罩效果&#xff0c;移动&#xff0c;关闭&#xff0c;最大化&#xff0c;最小化&#xff0c;还原操作 using System; using System.Collections.Generic; using System.ComponentModel; using System.Data; using System.Drawing; using System.Drawing.Drawing2D…
阅读更多...
在程序中链接静态库 和 动态库

在程序中链接静态库 和 动态库

9. 链接库 在编写程序的过程中&#xff0c;可能会用到一些系统提供的动态库或者自己制作出的动态库 或者静态库文件&#xff0c;cmake中也为我们提供了相关的加载动态库的命令hehedalinux:~/Linux/loveDBTeacher-v3$ tree . ├── CMakeLists.txt ├── include │ └── …
阅读更多...
如何从 Keras 中的深度学习目录加载大型数据集

如何从 Keras 中的深度学习目录加载大型数据集

一、说明 数据集读取&#xff0c;使用、在磁盘上存储和构建图像数据集有一些约定&#xff0c;以便在训练和评估深度学习模型时能够快速高效地加载。本文介绍Keras 深度学习库中的ImageDataGenerator类等工具自动加载训练、测试和验证数据集。 二、ImageDataGenerator加载数据集…
阅读更多...
中科院自动化所：基于关系图深度强化学习的机器人多目标包围问题新算法

中科院自动化所：基于关系图深度强化学习的机器人多目标包围问题新算法

摘要&#xff1a;中科院自动化所蒲志强教授团队&#xff0c;提出一种基于关系图的深度强化学习方法&#xff0c;应用于多目标避碰包围(MECA)问题&#xff0c;使用NOKOV度量动作捕捉系统获取多机器人位置信息&#xff0c;验证了方法的有效性和适应性。研究成果在2022年ICRA大会发…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023