【README】

本文记录了 kafka生产者开发方式；

【1】生产者概览

【1.1】kafka发送消息过程

【1.2】创建kafka生产者

1）创建kafka生产者，有3个必选属性：

bootstrap.servers： kakfa集群节点地址；
key.serializer：键序列化器；
value.serializer：值序列化器；

/* 1.创建kafka生产者的配置信息 */
Properties props = new Properties();
/* 指定连接的kafka集群, broker-list */
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "centos201:9092");
/* key, value 的序列化类 */
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
/** 设置压缩算法 */
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");/* 2.创建生产者对象 */
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

【2】发送消息到kafka

1）发送消息有3种方式：

发送并忘记：把消息发送给服务器，不管它是否到达；
同步发送：调用send()方法，返回一个Future对象，调用其get() 方法进入阻塞，服务器响应时，阻塞线程被唤醒并获得消息写入的元数据；
异步发送：调用send() 方法，并指定一个回调函数，服务器在响应是调用该函数；

【2.1】同步发送

/*** @Description 同步发送生产者* @author xiao tang* @version 1.0.0* @createTime 2021年12月09日*/
public class MyProducerSync {public static void main(String[] args) {// 1.创建kafka生产者的配置信息Properties props = new Properties();// 指定连接的kafka集群, broker-listprops.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.163.201:9092,192.168.163.202:9092,192.168.163.203:9092");// key, value 的序列化类props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 2.创建生产者对象KafkaProducer<String, String> producer = new KafkaProducer<>(props);// 3.发送数据Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>("hello10","k1", "v1"));try {// 当前线程阻塞，直到kafka响应返回写入消息的元数据RecordMetadata respMetadata = future.get();System.out.println("[生产者写入消息] 分区【" + respMetadata.partition() + "】-offset【" + respMetadata.offset() + "】");} catch (Exception e) {}// 关闭生产者producer.close();System.out.println("kafka生产者写入数据完成");}
}

kafka生产者一般发生两类错误：

可重试错误，如连接错误（通过再次建立连接来解决），无主错误（通过重新分区选举首领解决）；
不可重试错误，如消息太大错误；

【2.2】异步发送消息（带回调函数）

/*** @Description 【异步】发送生产者* @author xiao tang* @version 1.0.0* @createTime 2021年12月09日*/
public class MyProducerAsync {public static void main(String[] args) {// 1.创建kafka生产者的配置信息Properties props = new Properties();// 指定连接的kafka集群, broker-listprops.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.163.201:9092,192.168.163.202:9092,192.168.163.203:9092");// key, value 的序列化类props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 2.创建生产者对象KafkaProducer<String, String> producer = new KafkaProducer<>(props);// 3.发送数据producer.send(new ProducerRecord<String, String>("hello10","k1", "v1"),
new MyProducerCallback());// 关闭生产者producer.close();System.out.println("kafka生产者写入数据完成");}/*** @Description 生产者发送消息后回调类* @author xiao tang* @version 1.0.0* @createTime 2021年12月09日*/private static class MyProducerCallback implements Callback {// kafka服务器响应时回调方法@Overridepublic void onCompletion(RecordMetadata metadata, Exception exception) {if (exception == null) {System.out.println("[生产者写入消息成功] 分区【" + metadata.partition() + "】-offset【" + metadata.offset() + "】");} else {System.out.printf("写入kafka失败，异常信息【%s】", exception);}}}
}

【2.3】生产者配置

1）acks：有3个可选项；表示生产者消息被认为写入成功时，需要写入的副本个数；

0：无需判断，只要把消息发送出去，就认为成功；
1：仅首领副本；
all：所有副本；

2）buffer.memory： 设置生产者内存缓冲区大小，用于缓冲发送到服务器的消息；

若缓冲区不足，send() 方法要么阻塞，要么抛出异常；取决于如何设置 max.block.ms 参数（抛出异常前可以阻塞一段时间）；

3）compression.type：压缩算法；

默认不压缩；可选压缩算法包括 snappy, gzip ，lz4 ；
使用压缩可以降低网络传输开销和存储开销，这是 kafka发送消息的瓶颈所在；

4）retries： 发送消息失败时，生产者可以重试的次数；

如果达到这个次数，生产者会放弃重试并返回错误；默认情况下，生产者会在每次重试之间等待 100ms，通过 retry.backoff.ms 参数来改变这个时间间隔；
一般情况下，没必须处理可重试错误。但需要处理不可重试错误或重试次数超过上限的情况；

5）batch.size： 生产者把多个消息放在同一个批次里；该参数指定了一个批次可以使用的内存大小，单位字节；不过生产者不一定等到批次被填满才发送（参考 linger.ms）；

6）linger.ms：指定生产者在发送批次前等待更多消息加入批次的时间；

生产者会在批次填满或linger.ms 达到上限时把批次发送出去；
- 建议把linger.ms 设置为大于0的数，虽然增加了延时但提高了吞吐量；

7）client.id ： 任意字符串，服务器用它识别消息来源，还可以用在日志和配额指标里；

8）max.in.flight.requests.per.connection ： 指定生产者在收到服务器响应前可以发送多少个消息；

把它设置为1，可以保证消息是按照顺序写入服务器的，即使发生了重试；

9）timeout.ms , request.timeout.ms 和 metadata.fetch.timeout.ms

request.timeout.ms： 指定了生产者在发送数据时等待服务器返回响应的时间；
metadata.fetch.timeout.ms： 指定了生产者在获取元数据时等待服务器返回响应时间；若等待超时，要么重试，要么抛出异常；
timeout.ms：指定了broker等待同步副本返回消息确认的时间，与 acks 相匹配；

10）max.block.ms ： send() 方法或使用 partitionFor() 获取元数据时生产者的阻塞时间；

当生产者发送缓冲区已满，或没有可用的元数据，这些方法就会阻塞；在阻塞时间达到该值时，生产者抛出超时异常；

11）max.request.size： 指定生产者发送的请求大小；

可以指单个消息的最大值，也可以指单个请求所有消息总大小（如一批多个消息但走了一个请求）；
注意： broker对可接受的消息最大值有自己的限制（通过 message.max.bytes）指定；

12）receive.buffer.bytes 和 send.buffer.bytes

分别指定 TCP socket接收和发送数据包的缓冲区大小；如果设置为-1，使用操作系统默认值；

【2.4】生产者常用配置代码示例

public class MyProducer {public static void main(String[] args) {/* 1.创建kafka生产者的配置信息 */Properties props = new Properties();/*2.指定连接的kafka集群, broker-list */props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "centos201:9092");/*3.ack应答级别*/props.put(ProducerConfig.ACKS_CONFIG, "all");/*4.重试次数*/props.put(ProducerConfig.RETRIES_CONFIG, 0);/*5.批次大小，一次发送多少数据，当数据大于16k，生产者会发送数据到 kafka集群 */props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16 * KfkNumConst._1K);/*6.等待时间， 等待时间超过1毫秒，即便数据没有大于16k， 也会写数据到kafka集群 */props.put(ProducerConfig.LINGER_MS_CONFIG, 1);// 超时时间props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 3000);props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);/*7. RecordAccumulator 缓冲区大小*/props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 32 * KfkNumConst._1M);/*8. key, value 的序列化类 */props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());/** 设置压缩算法 */props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");System.out.println(props);/* 9.创建生产者对象 */KafkaProducer<String, String> producer = new KafkaProducer<>(props);/* 10.发送数据 */String now = DateUtils.getNowTimestamp();int order = 1;for (int i = 0; i < 50000; i++) {for (int j = 0; j < 3; j++) {Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>("hello10",j, "", String.format("[%s] ", order++) + now + " > " + DataFactory.INSTANCE.genChar(5)));try {System.out.println("[生产者] 分区【" + future.get().partition() + "】-offset【" + future.get().offset() + "】");} catch (Exception e) {}}}/* 11.关闭资源 */producer.close();System.out.println("kafka生产者写入数据完成");}
}

【3】分区

1）使用消息的键来做hash，以hash值作为分区号；

2）如果键为null，则使用默认分区器；默认使用轮询（Round Robin）算法把消息均衡分布到各个分区上；

【3.1】实现自定义分区策略

/*** @Description 自定义分区策略的生产者* @author xiao tang* @version 1.0.0* @createTime 2021年12月09日*/
public class MyProducerWithPartition {public static void main(String[] args) {// 1.创建kafka生产者的配置信息Properties props = new Properties();// 指定连接的kafka集群, broker-listprops.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.163.201:9092,192.168.163.202:9092,192.168.163.203:9092");// key, value 的序列化类props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 设置分区器props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName());// 2.创建生产者对象KafkaProducer<String, String> producer = new KafkaProducer<>(props);// 3.发送数据Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>("hello10","31", "v1"));try {// 当前线程阻塞，直到kafka响应返回写入消息的元数据RecordMetadata respMetadata = future.get();System.out.println("[生产者写入消息] 分区【" + respMetadata.partition() + "】-offset【" + respMetadata.offset() + "】");} catch (Exception e) {}// 关闭生产者producer.close();System.out.println("kafka生产者写入数据完成");}
}

分区器

/*** @Description 分区器* @author xiao tang* @version 1.0.0* @createTime 2021年12月09日*/
public class MyPartitioner implements Partitioner {// 对键首位字符ascii取分区数的模获得分区号@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {int partitionSize = cluster.partitionCountForTopic(topic);int operand = 0;if (key != null && String.valueOf(key).length() > 0) {operand = String.valueOf(key).codePointAt(0);}return operand % partitionSize;}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}

【4】拦截器

定义拦截器，设置拦截器属性（可配置多个拦截器）；

/** 设置拦截器 */
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
Arrays.asList(TimeInterceptor.class.getName()));

/*** @Description 时间拦截器* @author xiao tang* @version 1.0.0* @createTime 2021年12月10日*/
public class TimeInterceptor implements ProducerInterceptor<String, String> {@Overridepublic ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {// 在消息被序列化以及计算分区前调用， 追加时间戳（偷梁换柱）return new ProducerRecord<>(record.topic(), record.partition(), record.key(), record.value() + "[TimeInterceptor]" + DateUtils.getNowTimestamp());}@Overridepublic void onAcknowledgement(RecordMetadata metadata, Exception exception) {// 在消息从 RecordAccumulator 成功发送到Kafka Broker之后，或者在发送过程中失败时调用// 写入数据库}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}

消费消息日志：

消费者-分区【0】offset【7774】 -> 2021-12-10 21:05:32--[1] > ABCDE[TimeInterceptor]2021-12-10 21:05:30
消费者-分区【1】offset【7644】 -> 2021-12-10 21:05:32--[2] > ABCDE[TimeInterceptor]2021-12-10 21:05:32
消费者-分区【2】offset【7626】 -> 2021-12-10 21:05:32--[3] > ABCDE[TimeInterceptor]2021-12-10 21:05:32

kafka生产者开发方式相关推荐

2021年大数据Kafka（十）：kafka生产者数据分发策略
全网最详细的大数据Kafka文章系列,强烈建议收藏加关注! 新文章都已经列出历史文章目录,帮助大家回顾前面的知识重点. 目录系列历史文章生产者数据分发策略策略一:用户指定了partition 策 ...
discard connection丢失数据_python kafka 生产者发送数据的三种方式
python kafka 生产者发送数据的三种方式发送方式同步发送发送数据耗时最长有发送数据的状态,不会丢失数据,数据可靠性高以同步的方式发送消息时,一条一条的发送,对每条消息返回的结果判断 ...
java实现Kafka生产者示例
使用java实现Kafka的生产者 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 3 ...
kafka生产者和消费者端的数据不一致
撸了今年阿里.头条和美团的面试,我有一个重要发现.......>>> kafka生产者生产30条数据,而消费者却不一定消费了30条数据,经过探索发现了main线程执行完成了而kafk ...
mysql作为kafka生产者_Kafka之生产者
[TOC] 从编程的角度而言,生产者就是负责向 Kafka 发送消息的应用程序.在 Kafka 的历史变迁中, 一共有两个大版本的生产者客户端: 第-个是于 Kafka开源之初使用 Scala语言编 ...
kafka消费者开发方式小结
[README] 1, 本文总结了 kafka消费者开发方式: 2, 本文使用的是最新的kafka版本 3.0.0: [1] kafka消费则 [1.1]消费者与消费者组 1)消费者: 应用程序需要创 ...
spring-kafka整合:DefaultKafkaProducerFactory默认kafka生产者工厂介绍
[README] 0,为啥要看 DefaultKafkaProducerFactory? 最近在基于 springboot 开发kafka模块,发现 kafakTemplate构造器传入了 Defa ...
java客户端作为kafka生产者测试
[README] 1.本文主要对 java客户端作为kafka 生产者进行测试, 消费者由 centos的kafka命令行线程扮演: 2.消息发送: kafka的生产者采用异步发送消息的方式,在消息发 ...
java作为kafka生产者实验及Expiring超时问题解决
[README] java作为生产者,centos 作为消费者: [1]生产者代码 -- pom.xml  <dependencies><depen ...

kafka生产者开发方式

【README】

【1】生产者概览

【1.1】kafka发送消息过程

【1.2】创建kafka生产者

【2】发送消息到kafka

【2.1】同步发送

【2.2】异步发送消息（带回调函数）

【2.3】生产者配置

【2.4】生产者常用配置代码示例

【3】分区

【3.1】实现自定义分区策略

【4】拦截器

kafka生产者开发方式相关推荐

最新文章

热门文章

kafka生产者开发方式

【README】

【1】生产者概览

【1.1】kafka发送消息过程

【1.2】创建kafka生产者

【2】发送消息到kafka

【2.1】同步发送

【2.2】异步发送消息 （带回调函数）

【2.3】生产者配置

【2.4】生产者常用配置代码示例

【3】分区

【3.1】实现自定义分区策略

【4】拦截器

kafka生产者开发方式相关推荐

最新文章

热门文章

【2.2】异步发送消息（带回调函数）