Kafka 生产者

8月前作者：Java_Xjc 分类：Toy博客阅读(34) 违法举报

这篇具有很好参考价值的文章主要介绍了Kafka 生产者。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

目录

一、kafka生产者原理

二、kafka异步发送

配置kafka

创建对象，发送数据

带回调函数的异步发送

同步发送

文章来源地址https://www.toymoban.com/news/detail-413572.html

三、kafka生产者分区

分区策略

指定分区：

指定key：

什么都不指定：

自定义分区器

四、生产者提高吞吐量

五、数据的可靠性

ACK应答级别

数据完全可靠条件

可靠性总结

代码中配置ACK

五、数据重复

幂等性

生产者事务

六、数据有序、乱序

数据有序

数据乱序

一、kafka生产者原理

main线程中创建生产者对象，然后调用send方法来发送数据，根据生产环境的需求来判断是否要增加拦截器（一般建议不用），因为是跨节点通信，所以要对数据进行序列化，用kafka自带的序列化器处理，再由分区器来规定数据发送到哪个分区，先发送到了一个缓存队列当中，队列大小是32M，其中每个批次默认大小是16k，达到16k后就会进入sender线程发送到kafka集群，或者等待时间到了也会自动发送，每一个broker节点接受一个队列的消息，发送到分区后，分区最多缓存5个请求，如果达到5个请求都没有ack应答的话，那么接下来消息就会发送到别的分区上。

通过Selector将底层链路进行打通进行发送数据，集群收到后进行一个副本的同步，同步完成后进行ack应答。

Kafka 生产者

应答成功后清理队列中的数据

Kafka 生产者

应答失败会进行retry重试

Kafka 生产者

二、kafka异步发送

Kafka 生产者

Kafka 生产者

创建Maven工程导入kafka客户端依赖

Kafka 生产者

导包

Kafka 生产者

配置kafka

Kafka 生产者

创建对象，发送数据

Kafka 生产者

带回调函数的异步发送

Kafka 生产者

同步发送

Kafka 生产者

Kafka 生产者

三、kafka生产者分区

Kafka 生产者

分区策略

Kafka 生产者

Kafka 生产者

Kafka 生产者

指定分区：

Kafka 生产者

指定key：

Kafka 生产者

什么都不指定：

Kafka 生产者

怎么把订单表的数据发送到kafka的指定分区中？

key 上写表名，表名的hashcode值一定会发送到同一个分区（生产环境通常将表名作为key）

自定义分区器

Kafka 生产者

Kafka 生产者

Kafka 生产者

Kafka 生产者

在生产者中关联自定义分区器

Kafka 生产者

在生产环境中可以过滤一些脏数据！

四、生产者提高吞吐量

Kafka 生产者

压缩类型：

Kafka 生产者

Kafka 生产者

五、数据的可靠性

ACK应答级别

0的时候，数据发过来还没落盘就应答，结果leader挂了导致了数据丢失。

Kafka 生产者

1的时候，数据发送过来，leader落盘后就会应答，生产者收到ack应答认为信息已经发送成功，随后就会清除掉队列中的消息，但是此时follwer可能还没完成同步，这个时候leader挂掉，就会有一个follwer成为新的leader，可是生产者已经认为信息发送成功从队列中清除了消息，这就导致了数据的丢失。

Kafka 生产者

-1（all）：leader收到消息，并且所有follwer都完成消息同步后返回ack应答

Kafka 生产者

Kafka 生产者

Kafka 生产者 follwer挂掉的话，等时间达到阈值还没向Leader发送通信请求或同步数据就会被踢出ISR，意味着这个follwer就不是有效副本了！

Kafka 生产者

上面几种都有数据丢失的风险，如何真正保护数据的可靠性呢？

数据完全可靠条件

Kafka 生产者

ack级别设置为-1，且至少1个leader+1个follwer，ISR里min.insync.replicas >= 2

可靠性总结

Kafka 生产者

代码中配置ACK

Kafka 生产者

五、数据重复

Kafka 生产者

幂等性

Kafka 生产者

Kafka 生产者

生产者事务

Kafka 生产者

Kafka 生产者

Kafka 生产者

Kafka 生产者

六、数据有序、乱序

数据有序

Kafka 生产者

数据乱序

Kafka 生产者

到了这里，关于Kafka 生产者的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：如若内容造成侵权/违法违规/事实不符，请点击违法举报进行投诉反馈，一经查实，立即删除！

分享到：

领支付宝红包赞助服务器费用

kafka入门(五)：kafka生产者发送消息

构建消息，即创建 ProduceRecord 对象。 (1) kafka发送消息，最常见的构造方法是： topic 表示主题， value 表示值。 (2) kafka发送消息指定key，ProducerRecord 的 key ，既可以作为消息的唯一id，也可以用来决定消息该被写到主题的哪个分区。拥有相同key 的消息，将被写到同一个分区。

2024年01月17日
浏览(30)
多图详解 kafka 生产者消息发送过程

生产者客户端代码 KafkaProducer 通过解析 producer.propeties 文件里面的属性来构造自己。例如：分区器、Key 和 Value 序列化器、拦截器、 RecordAccumulator消息累加器、元信息更新器、启动发送请求的后台线程生产者元信息更新器我们之前有讲过. 客户端都会保存集群的元信息,例如

2023年04月09日
浏览(29)
分布式 - 消息队列Kafka：Kafka生产者发送消息的方式

不管是把Kafka作为消息队列、消息总线还是数据存储平台，总是需要一个可以往Kafka写入数据的生产者、一个可以从Kafka读取数据的消费者，或者一个兼具两种角色的应用程序。 Kafka 生产者是指使用 Apache Kafka 消息系统的应用程序，它们负责将消息发送到 Kafka 集群中的一个或多

2024年02月13日
浏览(32)
分布式 - 消息队列Kafka：Kafka生产者发送消息的分区策略

01. Kafka 分区的作用分区的作用就是提供负载均衡的能力，或者说对数据进行分区的主要原因，就是为了实现系统的高伸缩性。不同的分区能够被放置到不同节点的机器上，而数据的读写操作也都是针对分区这个粒度而进行的，这样每个节点的机器都能独立地执行各自分区的

2024年02月13日
浏览(37)
分布式 - 消息队列Kafka：Kafka生产者发送消息的3种方式

不管是把Kafka作为消息队列、消息总线还是数据存储平台，总是需要一个可以往Kafka写入数据的生产者、一个可以从Kafka读取数据的消费者，或者一个兼具两种角色的应用程序。 Kafka 生产者是指使用 Apache Kafka 消息系统的应用程序，它们负责将消息发送到 Kafka 集群中的一个或多

2024年02月13日
浏览(36)
Kafka 入门到起飞系列 - 生产者发送消息流程解析

生产者通过 producerRecord 对象封装消息主题、消息的value（内容）、timestamp(时间戳)等生产者通过 send() 方法发送消息，send()方法会经过如下几步 1. 首先将消息交给拦截器（Interceptor）处理, 拦截器对生产者而言，对所有消息都是生效的，拦截器也支持链式编程（责任器链）的

2024年02月16日
浏览(33)
kafka服务端允许生产者发送最大消息体大小

server.properties中加上的message.max.bytes配置，我目前设置为5242880，即5MB，可以根据实际情况增大。在生产者端配置max.request.size，这是单个消息最大字节数，根据实际调整，max.request.size 必须小于 message.max.bytes 以及消费者的 max.partition.fetch.bytes。这样消息

2024年02月15日
浏览(36)
Kafka中的生产者如何处理消息发送失败的情况？

在Kafka中，生产者可以通过以下方式处理消息发送失败的情况：同步发送模式（Sync Mode）：在同步发送模式下，生产者发送消息后会阻塞等待服务器的响应。如果发送失败，生产者会抛出异常（例如 ProducerRecord 发送异常）或返回错误信息。开发者可以捕获异常并根据需要进行

2024年02月06日
浏览(31)
07、Kafka ------ 消息生产者（演示发送消息）和消息消费者（演示监听消息）

简单来说，就是一个数据项。 ▲ 消息就是 Kafka 所记录的数据节点，消息在 Kafka 中又被称为记录（record）或事件（event）。从存储上来看，消息就是存储在分区文件（有点类似于List）中的一个数据项，消息具有 key、value、时间戳和可选的元数据头。 ▲ 下面是一个示例事件

2024年01月20日
浏览(30)
kafka生产者发送消息报错 Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected

报这个错误是因为kafka里的配置要修改下在config目录下 server.properties配置文件这下发送消息就不会一直等待，就可以发送成功了

2024年02月06日
浏览(26)