189 8069 5689

怎么解析KafkaProducer

本篇文章为大家展示了怎么解析Kafka Producer，内容简明扼要并且容易理解，绝对能使你眼前一亮，通过这篇文章的详细介绍希望你能有所收获。

创新互联建站是一家集网站建设,岳阳楼企业网站建设,岳阳楼品牌网站建设,网站定制,岳阳楼网站建设报价,网络营销,网络优化,岳阳楼网站推广为一体的创新建站企业，帮助传统企业提升企业形象加强企业竞争力。可充分满足这一群体相比中小企业更为丰富、高端、多元的互联网需求。同时我们时刻保持专业、时尚、前沿，时刻以成就客户成长自我，坚持不断学习、思考、沉淀、净化自己，让我们为更多的企业打造出实用型网站。

怎么解析Kafka Producer

Kafka Producer在发送消息大致有以下流程：

  1. 首先将消息封装在ProducerRecord中，并且序列化

  2. 将序列化后的消息发送给partitioner，partitioner主要用来确定消息发往哪个分区，默认的分区策略是轮询，如果消息有key，具有相同key的消息可以被发往同一分区，Kafka Producer也允许用户直接指定要发往的分区

  3. Producer有一个专门的Sender线程会从缓冲区获取消息，计算将要发往的Leader Broker然后进行发送。

必需参数
bootstrap.servers

指定一组host:port键值对，用于连接kafka broker节点，producer可以通过该参数发现Kafka集群中的所有broker，因此可以指定部分节点。

key.serializer

发往kafka的的消息都需要被序列化成二进制字节数组，该参数指定的是消息key的序列化格式，如果指定的StringSerializer，表明是将一个String字符串序列化成二进制字节数组。

目前支持的序列化格式有：

  • ByteArraySerializer：啥都不用做，本质就是个二进制字节数组

  • ByteBufferSerializer：序列化ByteBuffer

  • ByteSerializer：序列化Kafka自定义的Bytes类型

  • DoubleSerializer：序列化Double类型

  • IntegerSerializer：序列化Integer类型

  • LongSerializer：序列化Long类型

  • StringSerializer：序列化String类型

Kafka支持自定义序列化类型，只需要实现org.apache.kafka.common.serialization.Serializer接口。

value.serializer

指定消息内容的序列化格式

主要参数
acks

该参数主要用来控制消息的持久性，该参数主要有三个值：0、1、all(-1)

  • 0：表示Producer完全不用考虑broker的响应，发送完一条消息后立即开启下一条消息的发送，不用理会leader broker的响应，该设置吞吐量很高，但消息容易丢失

  • all(-1)：表示发送消息时，leader broker不仅会将消息写入本地日志，同时还会等待ISR中所有副本(ISR副本数量由broker端的min.insync.replicas控制)都成功写入他们的本地日志后，才发送响应结果给producer，该设置吞吐量最低，但消息持久性最高

  • 1：producer发送消息后，leader broker进将消息成功写入本地日志后便返回响应给producer

怎么解析Kafka Producer

buffer.memory

该参数用于指定producer发送消息缓冲区的大小，单位字节，默认值32MB。增加该参数的值可以提升吞吐量，但会增加延时。

compression.type

该参数用来设置是否开启消息压缩，默认值为none，目前Kafka支持GZIP、Snappy和LZ4。开启压缩后，可以降低网络I/O从而提升吞吐量，同时会增加CPU的压力。

retries

Producer在发送消息的时候有可能因为网络抖动从而发送失败，这种失败都是可以重试解决，retries参数决定了Producer内部的重试次数。

目前可以重试的异常主要有以下几种：

  • LeaderNotAvailableException：分区副本Leader不可用，通常出现在Leader选举期间

  • NotControllerException：controller不可用，通常出现在controller选举期间

  • NetworkException：网络瞬时故障导致的异常

重试有可能会导致以下问题：

  • 消息重复发送：kafka 0.11.0.0以前的版本需要用户对consumer进行去重处理，但在此版本以后可以在broker端将参数enable.idempotence设置为true，便可以实现消息的去重，此参数可以保证消息的幂等性

  • 消息乱序：Producer提供了max.in.flight.request.per.connect可以将该参数设置为1，表示Producer同一时刻只能发送一个请求

batch.size

Producer会将发往相同分区的消息进行汇总，当到达的该参数的设置值时，便将消息一同发往Leader Broker。

该参数的默认值是16KB，增加该参数可以提高吞吐量，但是相应的会增加消息的延迟

linger.ms

当消息无法满足batch size，但是超过了linger.ms的值时，也会将消息一同发往Leader Broker。

该参数默认值是0，表示不关心是否填满batch.size立即发送。

max.request.size

用于控制Producer发送请求的大小，默认值是1048576字节

request.timeout.ms

Producer发送请求给broker以后，broker需要在规定时间内返回响应，否则Producer会抛出TimemoutException需要客户端去处理。

该参数的默认值是30s。

多线程处理

KafkaProducer是线程安全的。

用法说明优势劣势
多线程单KafkaProducer实例所有线程共享一个KafkaProducer实例实现简单、性能好所有线程共享一个内存缓冲区，需要较多内存；一旦某个Producer线程崩溃将会导致KafkaProducer实例被破坏，所有线程无法工作
多线程多KafkaProducer实例每个线程维护自己的专属KafkaProducer实例每个用户线程拥有自己的专属KafkaProucer实例、缓冲区空间及一组对应的参数配置；单个KafkaProducer崩溃不会影响其他KafkaProducer需要较大内存开销

上述内容就是怎么解析Kafka Producer，你们学到知识或技能了吗？如果还想学到更多技能或者丰富自己的知识储备，欢迎关注创新互联行业资讯频道。


文章题目：怎么解析KafkaProducer
链接地址：http://cdxtjz.cn/article/pspshj.html

其他资讯