Kafka Producer在发送消息大致有以下流程:
指定一组host:port键值对,用于连接kafka broker节点,producer可以通过该参数发现Kafka集群中的所有broker,因此可以指定部分节点。
发往kafka的的消息都需要被序列化成二进制字节数组,该参数指定的是消息key的序列化格式,如果指定的StringSerializer,表明是将一个String字符串序列化成二进制字节数组。
目前支持的序列化格式有:
Kafka支持自定义序列化类型,只需要实现org.apache.kafka.common.serialization.Serializer接口。
指定消息内容的序列化格式
该参数主要用来控制消息的持久性,该参数主要有三个值:0、1、all(-1)
该参数用于指定producer发送消息缓冲区的大小,单位字节,默认值32MB。增加该参数的值可以提升吞吐量,但会增加延时。
该参数用来设置是否开启消息压缩,默认值为none,目前Kafka支持GZIP、Snappy和LZ4。开启压缩后,可以降低网络I/O从而提升吞吐量,同时会增加CPU的压力。
Producer在发送消息的时候有可能因为网络抖动从而发送失败,这种失败都是可以重试解决,retries参数决定了Producer内部的重试次数。
目前可以重试的异常主要有以下几种:
重试有可能会导致以下问题:
Producer会将发往相同分区的消息进行汇总,当到达的该参数的设置值时,便将消息一同发往Leader Broker。
该参数的默认值是16KB,增加该参数可以提高吞吐量,但是相应的会增加消息的延迟
当消息无法满足batch size,但是超过了linger.ms的值时,也会将消息一同发往Leader Broker。
该参数默认值是0,表示不关心是否填满batch.size立即发送。
用于控制Producer发送请求的大小,默认值是1048576字节
Producer发送请求给broker以后,broker需要在规定时间内返回响应,否则Producer会抛出TimemoutException需要客户端去处理。
该参数的默认值是30s。
KafkaProducer是线程安全的。
用法 | 说明 | 优势 | 劣势 |
---|---|---|---|
多线程单KafkaProducer实例 | 所有线程共享一个KafkaProducer实例 | 实现简单、性能好 | 所有线程共享一个内存缓冲区,需要较多内存;一旦某个Producer线程崩溃将会导致KafkaProducer实例被破坏,所有线程无法工作 |
多线程多KafkaProducer实例 | 每个线程维护自己的专属KafkaProducer实例 | 每个用户线程拥有自己的专属KafkaProucer实例、缓冲区空间及一组对应的参数配置;单个KafkaProducer崩溃不会影响其他KafkaProducer | 需要较大内存开销 |