Kafka系列（二）将消息数据写入Kafka系统--生产者【异步发送、同步发送、单线程发送、多线程发送、配置生产者属性、自定义序列化、自定义主题分区】

Kafka系列

    • 发送消息到 Kafka 主题
    • 了解异步模式
    • 了解同步模式
    • 线程发送消息的步骤
      • 生产者用单线程发送消息
      • 生产者用多线程发送消息
    • 配置生产者属性
    • 保存对象的各个属性一序列化
      • 序列化一个对象
      • 序列化对象的存储格式
      • 自己实现 序列化的步骤
        • 1. 创建序列化对象
        • 2. 编写序列化工具类
        • 3. 编写自定义序列化逻辑代码
        • 4. 编写生产者应用程序
    • 自定义主题分区
      • 编写自定义主题分区算法
      • 演示自定义分区的作用

转自 《Kafka并不难学！入门、进阶、商业实战》

发送消息到 Kafka 主题

Kafka 0.10.0.0 及以后的版本,对生产者代码的底层实现进行了重构。kafka.producer.Producer类被 org.apache.kafka.clients.producer.KafkaProducer 类替换
Kafka 系统支持两种不同的发送方式–同步模式(Sync)和异步模式(ASync)。

了解异步模式

在 Kafka 0.10.0.0 及以后的版本中，客户端应用程序调用生产者应用接口，默认使用异步的方式发送消息。
生产者客户端在通过异步模式发送消息时，通常会调用回调函数的 send（）方法发送消息。生产者端收到 Kafka 代理节点的响应后会触发回调函数。

  1. 什么场景下需要使用异步模式
    假如生产者客户端与 Kafka 集群节点间存在网络延时(100ms),此时发送 10 条消息记录,则延时将达到 1s。而大数据场景下有着海量的消息记录，发送的消息记录是远不止 10条，延时将非常严重。
    大数据场景下，如果采用异步模式发送消息记录，几乎没有任何耗时，通过回调函数可以知道消息发送的结果。
  2. 异步模式数据写入流程
    例如，一个业务主题(ip login)有6个分区。生产者客户端写入一条消息记录时，消息记录会先写入某个缓冲区，生产者客户端直接得到结果(这时，缓冲区里的数据并没有写到 Kafka代理节点中主题的某个分区)。之后，缓冲区中的数据会通过异步模式发送到 Kafka 代理节点中主题的某个分区中。
	//实例化一个消息记录对象，用来保存主题名、分区索引、键、值和时间戳ProducerRecord<byte[],byte[]> record =new ProducerRecord<byte[],byte[]>("ip login", key, value);//调用 send()方法和回调函数producer.send(myRecord,new Callback() {public void onCompletion (RecordMetadata metadata, Exception e){if (e != null) {e.printStackTrace();} else {System.out.println("The offset of the record we just sent is:" + metadata.offset());}}};

消息记录提交给 send()方法后，实际上该消息记录被放入一个缓冲区的发送队列，然后通过后台线程将其从缓冲区队列中取出并进行发送;发送成功后会触发send方法的回调函数-Callback.

了解同步模式

生产者客户端通过 send()方法实现同步模式发送消息，并返回一个 Future 对象，同时调用get()方法等待 Future 对象，看 send()方法是否发送成功。

  1. 什么场景下使用同步模式
    如果要收集用户访问网页的数据，在写数据到 Kafka 集群代理节点时需要立即知道消息是否写入成功，此时应使用同步模式。
// 将字符串转换成字节数组
byte[] key = "key".getBytes();
byte[] value ="value".getBytes();
// 实例化一个消息记录对象，用来保存主题名、分区索引、键、值和时间戳
ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("ip_login",key, value);
//调用 send()函数后，再通过 get()方法等待返回结果
producer.send(record).get();

这里通过调用 Future 接口中的 get()方法等待 Kafka 集群代理节点(Broker)的状态返回如果 Producer 发送消息记录成功了，则返回 RecordMetadata 对象，该对象可用来查看消息记录的偏移量(Offset)。

线程发送消息的步骤

在 Kafka 系统中，为了保证生产者客户端应用程序的独立运行，通常使用线程的方式发送消息。
创建一个简单的生产者应用程序的步骤如下。
(1)实例化 Properties 类对象，配置生产者应答机制。有以下三个属性是必须设置的。其他属性一般都会有默认值，可以按需添加设置。

  • bootstrap.servers:配置Kafka集群代理节点地址;
  • key.serializer:序列化消息主键;
  • value.serializer:序列化消息数据内容,

(2)根据属性对象实例化一个 KafkaProducer.
(3)通过实例化一个ProducerRecord 对象，将消息记录以“键-值”对的形式进行封装。
(4)通过调用 KafkaProducer 对象中带有回调函数的 send方法发送消息给 Kafka 集群。
(5)关闭KafkaProducer 对象，释放连接资源,

生产者用单线程发送消息

/*** 实现一个生产者客户端应用程序.*/
public class JProducer extends Thread {private final Logger LOG = LoggerFactory.getLogger(JProducer.class);/** 配置Kafka连接信息. */public Properties configure() {Properties props = new Properties();props.put("bootstrap.servers", "dn1:9092,dn2:9092,dn3:9092");// 指定Kafka集群地址props.put("acks", "1"); // 设置应答模式, 1表示有一个Kafka代理节点返回结果props.put("retries", 0); // 重试次数props.put("batch.size", 16384); // 批量提交大小props.put("linger.ms", 1); // 延时提交props.put("buffer.memory", 33554432); // 缓冲大小props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 序列化主键props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 序列化值return props;}public static void main(String[] args) {JProducer producer = new JProducer();producer.start();}/** 实现一个单线程生产者客户端. */public void run() {Producer<String, String> producer = new KafkaProducer<>(configure());// 发送100条JSON格式的数据for (int i = 0; i < 100; i++) {// 封装JSON格式JSONObject json = new JSONObject();json.put("id", i);json.put("ip", "192.168.0." + i);json.put("date", new Date().toString());String k = "key" + i;// 异步发送producer.send(new ProducerRecord<String, String>("test_kafka_game_x", k, json.toJSONString()), new Callback() {public void onCompletion(RecordMetadata metadata, Exception e) {if (e != null) {LOG.error("Send error, msg is " + e.getMessage());} else {LOG.info("The offset of the record we just sent is: " + metadata.offset());}}});}try {sleep(3000);// 间隔3秒} catch (InterruptedException e) {LOG.error("Interrupted thread error, msg is " + e.getMessage());}producer.close();// 关闭生产者对象}
}

这里的主题只有一个分区和一个副本，所以，发送的所有消息会写入同一个分区中
如果希望发送完消息后获取一些返回信息(比如获取消息的偏移量、分区索引值、提交的时间戳等)，则可以通过回调函数 CallBack 返回的 RecordMetadata 对象来实现。
由于 Kafka 系统的生产者对象是线程安全的，所以，可通过增加生产者对象的线程数来提高 Kafka 系统的吞吐量。

生产者用多线程发送消息

public class JProducerThread extends Thread {// 创建一个日志对象private final Logger LOG = LoggerFactory.getLogger(JProducerThread.class);// 声明最大线程数private final static int MAX_THREAD_SIZE = 6;/** 配置Kafka连接信息. */public Properties configure() {Properties props = new Properties();props.put("bootstrap.servers", "dn1:9092,dn2:9092,dn3:9092");// 指定Kafka集群地址props.put("acks", "1"); // 设置应答模式, 1表示有一个Kafka代理节点返回结果props.put("retries", 0); // 重试次数props.put("batch.size", 16384); // 批量提交大小props.put("linger.ms", 1); // 延时提交props.put("buffer.memory", 33554432); // 缓冲大小props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 序列化主键props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 序列化值props.put("partitioner.class", "org.smartloli.kafka.game.x.book_4.JPartitioner");// 指定自定义分区类return props;}public static void main(String[] args) {// 创建一个固定线程数量的线程池ExecutorService executorService = Executors.newFixedThreadPool(MAX_THREAD_SIZE);// 提交任务executorService.submit(new JProducerThread());// 关闭线程池executorService.shutdown();}/** 实现一个单线程生产者客户端. */public void run() {Producer<String, String> producer = new KafkaProducer<>(configure());// 发送100条JSON格式的数据for (int i = 0; i < 10; i++) {// 封装JSON格式JSONObject json = new JSONObject();json.put("id", i);json.put("ip", "192.168.0." + i);json.put("date", new Date().toString());String k = "key" + i;// 异步发送producer.send(new ProducerRecord<String, String>("ip_login_rt", k, json.toJSONString()), new Callback() {public void onCompletion(RecordMetadata metadata, Exception e) {if (e != null) {LOG.error("Send error, msg is " + e.getMessage());} else {LOG.info("The offset of the record we just sent is: " + metadata.offset());}}});}try {sleep(3000);// 间隔3秒} catch (InterruptedException e) {LOG.error("Interrupted thread error, msg is " + e.getMessage());}producer.close();// 关闭生产者对象}}

配置生产者属性

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

保存对象的各个属性一序列化

序列化一个对象

在分布式环境下，无论哪种格式的数据都会被分解成二进制，以便存储在文件中或者在网络上传输。
序列化就是，将对象以一连串的字节进行描述，用来解决对象在进行读写操作时所引发的问题。
序列化可以将对象的状态写成数据流，并进行网络传输或者保存在文件或数据库中，在需要时再把该数据流读取出来，重新构造一个相同的对象。

  1. 为什么需要序列化
    在传统的企业应用中，不同的组件分布在不同的系统和网络中。如果两个组件之间想要进行通信，那么它们之间必须有数据转换机制。实现这个过程需要遵照一个协议来传输对象，这意味着，接收端需要知道发送端所使用的协议才能重新构建对象，以此来保证两个组件之间的通信是安全的。
public class JObjectSerial implements Serializable {private static Logger LOG = LoggerFactory.getLogger(JObjectSerial.class);/*** 序列化版本ID.*/private static final long serialVersionUID = 1L;public byte id = 1; // 用户IDpublic byte money = 100; // 充值金额/** 实例化入口函数. */public static void main(String[] args) {try {FileOutputStream fos = new FileOutputStream("/tmp/salary.out"); // 实例化一个输出流对象ObjectOutputStream oos = new ObjectOutputStream(fos);// 实例化一个对象输出流JObjectSerial jos = new JObjectSerial(); // 实例化序列化类oos.writeObject(jos); // 写入对象oos.flush(); // 刷新数据流oos.close();// 关闭连接} catch (Exception e) {LOG.error("Serial has error, msg is " + e.getMessage());// 打印异常信息}}
}

序列化对象的存储格式

在这里插入图片描述

自己实现 序列化的步骤

在这里插入图片描述
如果使用原生的序列化方式，则需要将传输的内容拼接成字符串或转成字符数组，抑或是其他类型，这样在实现代码时就会比较麻烦。而 Kafka 为了解决这种问题，提供了序列化的接口，让用户可以自定义对象的序列化方式，来完成对象的传输。
以下实例将演示生产者客户端应用程序中序列化的用法，利用 Serializable 接口来序列化对象。

1. 创建序列化对象
/*** 声明一个序列化类.* * @author smartloli.**         Created by Apr 30, 2018*/
public class JSalarySerial implements Serializable {/*** 序列化版本ID.*/private static final long serialVersionUID = 1L;private String id;// 用户IDprivate String salary;// 金额public String getId() {return id;}public void setId(String id) {this.id = id;}public String getSalary() {return salary;}public void setSalary(String salary) {this.salary = salary;}// 打印对象属性值@Overridepublic String toString() {return "JSalarySerial [id=" + id + ", salary=" + salary + "]";}}
2. 编写序列化工具类
/*** 封装一个序列化的工具类.* * @author smartloli.**         Created by Apr 30, 2018*/
public class SerializeUtils {/** 实现序列化. */public static byte[] serialize(Object object) {try {return object.toString().getBytes("UTF8");// 返回字节数组} catch (Exception e) {e.printStackTrace(); // 抛出异常信息}return null;}/** 实现反序列化. */public static <T> Object deserialize(byte[] bytes) {try {return new String(bytes, "UTF8");// 反序列化} catch (Exception e) {e.printStackTrace();}return null;}}
3. 编写自定义序列化逻辑代码
/*** 自定义序列化实现.* * @author smartloli.**         Created by Apr 30, 2018*/
public class JSalarySeralizer implements Serializer<JSalarySerial> {@Overridepublic void configure(Map<String, ?> configs, boolean isKey) {}/** 实现自定义序列化. */@Overridepublic byte[] serialize(String topic, JSalarySerial data) {return SerializeUtils.serialize(data);}@Overridepublic void close() {}}
4. 编写生产者应用程序
/*** 自定义序列化, 发送消息给Kafka.* * @author smartloli.**         Created by Apr 30, 2018*/
public class JProducerSerial extends Thread {private static Logger LOG = LoggerFactory.getLogger(JProducerSerial.class);/** 配置Kafka连接信息. */public Properties configure() {Properties props = new Properties();props.put("bootstrap.servers", "dn1:9092,dn2:9092,dn3:9092");// 指定Kafka集群地址props.put("acks", "1"); // 设置应答模式, 1表示有一个Kafka代理节点返回结果props.put("retries", 0); // 重试次数props.put("batch.size", 16384); // 批量提交大小props.put("linger.ms", 1); // 延时提交props.put("buffer.memory", 33554432); // 缓冲大小props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 序列化主键props.put("value.serializer", "org.smartloli.kafka.game.x.book_4.serialization.JSalarySeralizer");// 自定义序列化值return props;}public static void main(String[] args) {JProducerSerial producer = new JProducerSerial();producer.start();}/** 实现一个单线程生产者客户端. */public void run() {Producer<String, JSalarySerial> producer = new KafkaProducer<>(configure());JSalarySerial jss = new JSalarySerial();jss.setId("2018");jss.setSalary("100");producer.send(new ProducerRecord<String, JSalarySerial>("test_topic_ser_des", "key", jss), new Callback() {public void onCompletion(RecordMetadata metadata, Exception e) {if (e != null) {LOG.error("Send error, msg is " + e.getMessage());} else {LOG.info("The offset of the record we just sent is: " + metadata.offset());}}});try {sleep(3000);// 间隔3秒} catch (InterruptedException e) {LOG.error("Interrupted thread error, msg is " + e.getMessage());}producer.close();// 关闭生产者对象}
}

在这里插入图片描述

自定义主题分区

在这里插入图片描述

编写自定义主题分区算法

/*** 实现一个自定义分区类.** @author smartloli.**         Created by Apr 30, 2018*/
public class JPartitioner implements Partitioner {@Overridepublic void configure(Map<String, ?> configs) {}/** 实现Kafka主题分区索引算法. */@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {int partition = 0;String k = (String) key;partition = Math.abs(k.hashCode()) % cluster.partitionCountForTopic(topic);return partition;}@Overridepublic void close() {}}

演示自定义分区的作用

/*** 实现一个生产者客户端应用程序.* * @author smartloli.**         Created by Apr 27, 2018*/
public class JProducerThread extends Thread {// 创建一个日志对象private final Logger LOG = LoggerFactory.getLogger(JProducerThread.class);// 声明最大线程数private final static int MAX_THREAD_SIZE = 6;/** 配置Kafka连接信息. */public Properties configure() {Properties props = new Properties();props.put("bootstrap.servers", "dn1:9092,dn2:9092,dn3:9092");// 指定Kafka集群地址props.put("acks", "1"); // 设置应答模式, 1表示有一个Kafka代理节点返回结果props.put("retries", 0); // 重试次数props.put("batch.size", 16384); // 批量提交大小props.put("linger.ms", 1); // 延时提交props.put("buffer.memory", 33554432); // 缓冲大小props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 序列化主键props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");// 序列化值props.put("partitioner.class", "org.smartloli.kafka.game.x.book_4.JPartitioner");// 指定自定义分区类return props;}public static void main(String[] args) {// 创建一个固定线程数量的线程池ExecutorService executorService = Executors.newFixedThreadPool(MAX_THREAD_SIZE);// 提交任务executorService.submit(new JProducerThread());// 关闭线程池executorService.shutdown();}/** 实现一个单线程生产者客户端. */public void run() {Producer<String, String> producer = new KafkaProducer<>(configure());// 发送100条JSON格式的数据for (int i = 0; i < 10; i++) {// 封装JSON格式JSONObject json = new JSONObject();json.put("id", i);json.put("ip", "192.168.0." + i);json.put("date", new Date().toString());String k = "key" + i;// 异步发送producer.send(new ProducerRecord<String, String>("ip_login_rt", k, json.toJSONString()), new Callback() {public void onCompletion(RecordMetadata metadata, Exception e) {if (e != null) {LOG.error("Send error, msg is " + e.getMessage());} else {LOG.info("The offset of the record we just sent is: " + metadata.offset());}}});}try {sleep(3000);// 间隔3秒} catch (InterruptedException e) {LOG.error("Interrupted thread error, msg is " + e.getMessage());}producer.close();// 关闭生产者对象}}

在这里插入图片描述

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

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

相关文章

[技术杂谈]如何下载vscode历史版本

[技术杂谈]如何下载vscode历史版本

网站模板&#xff1a; https://code.visualstudio.com/updates/v1_85 如果你想下载1.84系列可以访问https://code.visualstudio.com/updates/v1_84​​​​​​ 然后看到&#xff1a; 选择对应版本下载即可&#xff0c;我是windows x64系统选择x64即可开始下载
阅读更多...
Element UI+Spring Boot进行CRUD的实例

Element UI+Spring Boot进行CRUD的实例

ElementUI安装与使用指南 前端代码&#xff1a;点击查看learnelementuispringboot项目源码 后端代码&#xff1a;点击查看 LearnElementUiAndSpringBoot 一、前端配置 安装axios axios官网axios中文文档安装指令&#xff1a;npm install axios 二、后端配置 springboot3m…
阅读更多...
C#验证字符串的长度，用正则表达式 vs 字符数组长度或字符串的长度

C#验证字符串的长度，用正则表达式 vs 字符数组长度或字符串的长度

目录 一、使用的方法 1.使用正则表达式 2.通过计算字符串的长度验证 二、实例 1.源码 2.生成效果 一、使用的方法 1.使用正则表达式 使用正则表达式可以判断和限制用户输入的字符串长度。 比如验证用户密码不得少于8为&#xff0c;匹配的正则表达式"^.{8,}$"…
阅读更多...
电脑上常见的绘图软件有哪些？

电脑上常见的绘图软件有哪些？

现在在电脑上绘图很流行&#xff0c;不仅可以随时更改&#xff0c;还可以提高绘图效率&#xff0c;绘图软件中有很多工具。市场上的计算机绘图软件种类繁多。包括艺术设计、工业绘图和3D绘图。那么每个绘图软件都有自己的特点。那么&#xff0c;哪个更适合计算机绘画软件呢&…
阅读更多...
极速上手：使用Jmeter轻松实现N种参数化

极速上手：使用Jmeter轻松实现N种参数化

参数化的方式&#xff1a; 一、使用用户自定义变量 一种方式&#xff1a;直接在测试计划中添加用户自定义变量 另外一种方式&#xff1a;配置元件——用户自定义变量 示例&#xff1a;用户自定义变量&#xff0c;登录手机号码 在接口请求的时候&#xff0c;进行引用 请求之后&…
阅读更多...
PyTorch 2.2 中文官方教程（十一）

PyTorch 2.2 中文官方教程（十一）

使用 PyTorch C 前端 原文&#xff1a;pytorch.org/tutorials/advanced/cpp_frontend.html 译者&#xff1a;飞龙 协议&#xff1a;CC BY-NC-SA 4.0 PyTorch C 前端是 PyTorch 机器学习框架的纯 C 接口。虽然 PyTorch 的主要接口自然是 Python&#xff0c;但这个 Python API 坐…
阅读更多...
Qt程序设计-自定义QLineEdit控件添加鼠标单击事件

Qt程序设计-自定义QLineEdit控件添加鼠标单击事件

本文讲解Qt自定义QLineEdit控件添加鼠标单击事件。 QLineEdit控件默认没有单击事件，但是项目开发中有时需要单击事件，比如单击QLineEdit控件弹出软键盘。具体实现过程如下： 创建项目，在项目中添加一个类，命名为MyLineEdit 输入继承QLineEdit #ifndef MYLINEEDIT_H #defi…
阅读更多...
Sklearn、TensorFlow 与 Keras 机器学习实用指南第三版（七）

Sklearn、TensorFlow 与 Keras 机器学习实用指南第三版（七）

原文&#xff1a;Hands-On Machine Learning with Scikit-Learn, Keras, and TensorFlow 译者&#xff1a;飞龙 协议&#xff1a;CC BY-NC-SA 4.0 第十六章&#xff1a;使用 RNN 和注意力进行自然语言处理 当艾伦图灵在 1950 年想象他著名的Turing 测试时&#xff0c;他提出了…
阅读更多...
FL Studio Producer Edition 21.2.2全插件版+Crack下载链接（亲测可用,非钓鱼）

FL Studio Producer Edition 21.2.2全插件版+Crack下载链接（亲测可用,非钓鱼）

不知道为什么现在钓鱼的这么多&#xff08;有答案的请在评论区上告诉我&#xff09;&#xff0c;就一个学习版的编曲软件有必要这样子搞吗&#xff1f;我也是在各类博客上找了一大堆教程&#xff0c;根本没几个能用的&#xff0c;索性直接到海盗湾上找了一个&#xff0c;发现可…
阅读更多...
python 爬虫安装http请求库

python 爬虫安装http请求库

我的是window环境&#xff0c;安装的python3&#xff0c;如果再linux环境&#xff1a;pip install requests 开始&#xff1a; 上面我们成功发送请求并获取到响应&#xff0c;现在需要解析html或xml获取数据&#xff0c;因此我使用现成的工具库Beautiful Soup
阅读更多...
windows 搭建nginx http服务

windows 搭建nginx http服务

下载 下面链接直接点击下载&#xff0c;下载的就是包含rtmp服务器相关功能的&#xff0c;只不过需要配置下 Index of /download/ (ecsds.eu) nginx 1.7.11.3 Gryphon.zip直接点击额下面的连接即可下载 http://nginx-win.ecsds.eu/download/nginx%201.7.11.3%20Gryphon.zip …
阅读更多...
操作系统-【预备学习-1】（Linux 文件目录）

操作系统-【预备学习-1】（Linux 文件目录）

文章目录 相关知识目录结构进入目录补充查看目录创建文件删除文件创建文件夹删除文件夹文件和文件夹拷贝文件和文件夹移动/重命名 任务要求 相关知识 目录结构 Linux 文件系统是树形层次结构&#xff0c;具体如下图所示&#xff0c;最重要的是根目录&#xff08;/&#xff09…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023