东莞寮步汽车城,济宁seo优化公司,深圳网站制作比较好公司,wordpress+上一篇+下一篇理解Kafka正确使用方式
Kafka提供了两套客户端API#xff0c;HighLevel API和LowLevel API。 HighLevel API封装了kafka的运行细节#xff0c;使用起来比较简单#xff0c;是企业开发过程中最常用的客户端API。 LowLevel API则需要客户端自己管理Kafka的运行细节#xf…理解Kafka正确使用方式
Kafka提供了两套客户端APIHighLevel API和LowLevel API。 HighLevel API封装了kafka的运行细节使用起来比较简单是企业开发过程中最常用的客户端API。 LowLevel API则需要客户端自己管理Kafka的运行细节PartitionOffset这些数据都由客户端自行管理。这层API功能更灵活但是使用起来非常复杂也更容易出错。只在极少数对性能要求非常极致的场景才会偶尔使用。 基础的客户端
引入Maven依赖
dependencygroupIdorg.apache.kafka/groupIdartifactIdkafka_2.13/artifactIdversion3.4.0/version
/dependency
消息发送者主流程
public class MyProducer
{// private static final String BOOTSTRAP_SERVERS worker1:9092,worker2:9092,worker3:9092;private static final String BOOTSTRAP_SERVERS 你的公网IP:9092;private static final String TOPIC disTopic;public static void main(String[] args)throws ExecutionException, InterruptedException{// PART1:设置发送者相关属性Properties props new Properties();// 此处配置的是kafka的端口props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);// 配置key的序列化类props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);// 配置value的序列化类props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,org.apache.kafka.common.serialization.StringSerializer);ProducerString, String producer new KafkaProducer(props);CountDownLatch latch new CountDownLatch(5);for (int i 0; i 5; i){// Part2:构建消息ProducerRecordString, String record new ProducerRecord(TOPIC, Integer.toString(i), MyProducer i);// Part3:发送消息// 单向发送不关心服务端的应答。// producer.send(record);// System.out.println(message i sended);// 同步发送获取服务端应答消息前会阻塞当前线程。// RecordMetadata recordMetadata producer.send(record).get();// String topic recordMetadata.topic();// int partition recordMetadata.partition();// long offset recordMetadata.offset();// String message recordMetadata.toString();// System.out.println(message:[ message ] sended with topic: topic ; partition: partition// ;offset: offset);// 异步发送消息发送后不阻塞服务端有应答后会触发回调函数producer.send(record, new Callback(){Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e){if (null ! e){System.out.println(消息发送失败, e.getMessage());e.printStackTrace();}else{String topic recordMetadata.topic();long offset recordMetadata.offset();String message recordMetadata.toString();System.out.println(message:[ message ] sended with topic: topic ;offset: offset);}latch.countDown();}});}// 消息处理完才停止发送者。latch.await();producer.close();}
}
构建Producer分为三个步骤
1、设置Producer核心属性Producer可选的属性都可以由ProducerConfig类管理。比如ProducerConfig.BOOTSTRAP_SERVERS_CONFIG属性显然就是指发送者要将消息发到哪个Kafka集群上。这是每个Producer必选的属性。在ProducerConfig中对于大部分比较重要的属性都配置了对应的DOC属性进行描述。
2、构建消息Kafka的消息是一个Key-Value结构的消息。其中key和value都可以是任意对象类型。其中key主要是用来进行Partition分区的业务上更关心的是value。
3、使用Producer发送消息通常用到的就是单向发送、同步发送和异步发送者三种发送方式。 异常信息: Expiring 1 record(s) for disTopic-0:120010 ms has passed since batch creation
修改kafka安装路径的config/server.properties配置然后重启解决问题
# The address the socket server listens on. If not configured, the host name will be equal to the value of
# java.net.InetAddress.getCanonicalHostName(), with PLAINTEXT listener name, and port 9092.
# FORMAT:
# listeners listener_name://host_name:port
# EXAMPLE:
# listeners PLAINTEXT://your.host.name:9092
listenersPLAINTEXT://:9092# Listener name, hostname and port the broker will advertise to clients.
# If not set, it uses the value for listeners.
advertised.listenersPLAINTEXT://你的公网IP:9092
注意云服务器配置安全组开放端口虚拟机防火墙状态要关闭 消息消费者主流程
public class MyConsumer
{// private static final String BOOTSTRAP_SERVERS worker1:9092,worker2:9092,worker3:9092;private static final String BOOTSTRAP_SERVERS 你的公网IP:9092;private static final String TOPIC disTopic;public static void main(String[] args){// PART1:设置发送者相关属性Properties props new Properties();// kafka地址props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);// 每个消费者要指定一个groupprops.put(ConsumerConfig.GROUP_ID_CONFIG, test);// key序列化类props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,org.apache.kafka.common.serialization.StringDeserializer);// value序列化类props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,org.apache.kafka.common.serialization.StringDeserializer);ConsumerString, String consumer new KafkaConsumer(props);consumer.subscribe(Arrays.asList(TOPIC));// 自行调整Offset// consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC,0)));while (true){// PART2:拉取消息// 100毫秒超时时间ConsumerRecordsString, String records consumer.poll(Duration.ofNanos(100));// records.partitions().forEach(topicPartition - {// String key topicPartition.topic()topicPartition.partition();// ListConsumerRecordString, String partionRecords records.records(topicPartition);// long value partionRecords.get(partionRecords.size()-1).offset();//// });// PART3:处理消息for (ConsumerRecordString, String record : records){System.out.println(partition record.partition() offset record.offset() ;key record.key() ; value record.value());}// 提交offset消息就不会重复推送。consumer.commitSync(); // 同步提交表示必须等到offset提交完毕再去消费下一批数据。// consumer.commitAsync(); //异步提交表示发送完提交offset请求后就开始消费下一批数据了。不用等到Broker的确认。}}
}
Consumer同样是分为三个步骤
1、设置Consumer核心属性 可选的属性都可以由ConsumerConfig类管理。在这个类中同样对于大部分比较重要的属性都配置了对应的DOC属性进行描述。同样BOOTSTRAP_SERVERS_CONFIG是必须设置的属性。
2、拉取消息Kafka采用Consumer主动拉取消息的Pull模式。consumer主动从Broker上拉取一批感兴趣的消息。
3、处理消息提交位点消费者将消息拉取完成后就可以交由业务自行处理对应的这一批消息了。只是消费者需要向Broker提交偏移量offset。如果不提交OffsetBroker会认为消费者端消息处理失败了还会重复进行推送。 kafka官方配置 Apache Kafka 从客户端属性来梳理客户端工作机制
Kafka的设计精髓是在网络不稳定服务也随时会崩溃的这些作死的复杂场景下如何保证消息的高并发、高吞吐。首先要理解基础模型。
消费者分组消费机制
最重要的一个机制
在Consumer中都需要指定一个GROUP_ID_CONFIG属性表示当前Consumer所属的消费者组。
public static final String GROUP_ID_CONFIG group.id;
public static final String GROUP_ID_DOC A unique string that identifies the consumer group this consumer belongs to. This property is required if the consumer uses either the group management functionality by using codesubscribe(topic)/code or the Kafka-based offset management strategy.;//参数GROUP_INSTANCE_ID_CONFIG给组成员设置一个固定的instanceId可以减少Kafka不必要的rebalance
描述翻译对于Consumer如果需要在subcribe时使用组管理功能以及Kafka提供的offset管理策略那就必须要配置GROUP_ID_CONFIG属性。 消费者组作用生产者往Topic下发消息时会尽量均匀的将消息发送到Topic下的各个Partition当中。而这个消息会向所有订阅了该Topic的消费者推送。推送时每个ConsumerGroup中只会推送一份。也就是同一个消费者组中的多个消费者实例只会共同消费一个消息副本。而不同消费者组之间会重复消费消息副本。
Offset偏移量表示每个消费者组在每个Partiton中已经消费处理的进度
#查看消费者组Offset记录情况
bin/kafka-consumer-groups.sh --bootstrap-server worker1:9092 --describe --group test
这个Offset偏移量需要消费者处理完成后主动向Kafka的Broker提交。提交完成后Broker就会更新消费进度表示这个消息已经被这个消费者组处理完了。但是如果消费者没有提交OffsetBroker就会认为这个消息还没有被处理过就会重新往对应的消费者组进行推送不过这次一般会尽量推送给同一个消费者组当中的其他消费者实例。
Kafka自动提交
public static final String ENABLE_AUTO_COMMIT_CONFIG enable.auto.commit;
private static final String ENABLE_AUTO_COMMIT_DOC If true the consumers offset will be periodically committed in the background.;
如何提高Offset数据的安全性呢
服务端在Consumer中实际上提供了AUTO_OFFSET_RESET_CONFIG参数来指定消费者组在服务端的Offset不存在时如何进行后续消费。有可能服务端初始化Consumer Group的Offset失败也有可能Consumer Group当前的Offset对应的数据文件被过期删除了 ConsumerConfig.AUTO_OFFSET_RESEWT_CONFIG 当Server端没有对应的Offset时如何处理。 可选项 earliest 自动设置为当前最早的offset latest自动设置为当前最晚的offset none 如果消费者组对应的offset找不到就向Consumer抛异常。 其他选项 向Consumer抛异常。 客户端 异步提交消费者在处理业务的同时异步向Broker提交Offset。效率高但是容易丢数据处理失败但是offset提交了 同步提交消费者保证处理完所有业务后再提交Offset。消息不会因为offset丢失。业务处理失败不提交Offset还可以重试。坏处是处理变慢另外还可能重复消费生产者推送消息给组内的其他消费者 生产者拦截器机制
允许客户端在生产者消息发送到Kafka集群之前对消息进行拦截甚至可以修改消息内容
涉及到Producer中指定的一个参数INTERCEPTOR_CLASSES_CONFIG
public static final String INTERCEPTOR_CLASSES_CONFIG interceptor.classes;
public static final String INTERCEPTOR_CLASSES_DOC A list of classes to use as interceptors. Implementing the codeorg.apache.kafka.clients.producer.ProducerInterceptor/code interface allows you to intercept (and possibly mutate) the records received by the producer before they are published to the Kafka cluster. By default, there are no interceptors.;
按照说明定义一个自己的拦截器实现类
public class MyInterceptor implements ProducerInterceptor
{// 发送消息时触发Overridepublic ProducerRecord onSend(ProducerRecord producerRecord){System.out.println(prudocerRecord : producerRecord.toString());return producerRecord;}// 收到服务端响应时触发Overridepublic void onAcknowledgement(RecordMetadata recordMetadata, Exception e){System.out.println(acknowledgement recordMetadata: recordMetadata.toString());}// 连接关闭时触发Overridepublic void close(){System.out.println(producer closed);}// 整理配置项Overridepublic void configure(MapString, ? map){System.out.println(config start);for (Map.EntryString, ? entry : map.entrySet()){System.out.println(entry.key: entry.getKey() entry.value: entry.getValue());}System.out.println(config end);}
}
然后在生产者中指定拦截器类多个拦截器类用逗号隔开
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, com.gao.kfk.basic.MyInterceptor);
拦截器机制一般用得比较少主要用在一些统一添加时间等类似的业务场景。比如用Kafka传递一些POJO就可以用拦截器统一添加时间属性。但是我们平常用Kafka传递的都是String类型的消息POJO类型的消息Kafka可以传吗这就要用到下面的消息序列化机制。 消息序列化机制
Producer指定了两个属性KEY_SERIALIZER_CLASS_CONFIG和VALUE_SERIALIZER_CLASS_CONFIG对于这两个属性在ProducerConfig中都有配套的说明属性。
public static final String KEY_SERIALIZER_CLASS_CONFIG key.serializer;
public static final String KEY_SERIALIZER_CLASS_DOC Serializer class for key that implements the codeorg.apache.kafka.common.serialization.Serializer/code interface.;
public static final String VALUE_SERIALIZER_CLASS_CONFIG value.serializer;
public static final String VALUE_SERIALIZER_CLASS_DOC Serializer class for value that implements the codeorg.apache.kafka.common.serialization.Serializer/code interface.;
通过这两个参数可以指定消息生产者如何将消息的key和value序列化成二进制数据。 key是用来进行分区的可选项。Kafka通过key来判断消息要分发到哪个Partition。 如果没有填写key那么Kafka会使Round-robin轮询的方式自动选择Partition。 如果填写了key那么会通过声明的Serializer序列化接口将key转换成一个byte[]数组然后对key进行hash选择Partition。这样可以保证key相同的消息会分配到相同的Partition中。 Value是业务上比较关心的消息。Kafka同样需要将Value对象通过Serializer序列化接口将Key转换成byte[]数组这样才能比较好的在网络上传输Value信息以及将Value信息落盘到操作系统的文件当中。
生产者要对消息进行序列化那么消费者拉取消息时自然需要进行反序列化。所以在Consumer中也有反序列化的两个配置
public static final String KEY_DESERIALIZER_CLASS_CONFIG key.deserializer;
public static final String KEY_DESERIALIZER_CLASS_DOC Deserializer class for key that implements the codeorg.apache.kafka.common.serialization.Deserializer/code interface.;
public static final String VALUE_DESERIALIZER_CLASS_CONFIG value.deserializer;
public static final String VALUE_DESERIALIZER_CLASS_DOC Deserializer class for value that implements the codeorg.apache.kafka.common.serialization.Deserializer/code interface.;
在Kafka中对于常用的一些基础数据类型都已经提供了对应的实现类。但是如果需要使用一些自定义的消息格式比如自己定制的POJO就需要定制具体的实现类了。
在自己进行序列化机制时需要考虑的是如何用二进制来描述业务数据
一种类型是定长的基础类型比如Integer,Long,Double等。这些基础类型转化成二进制数组都是定长的。这类属性可以直接转成序列化数组在反序列化时只要按照定长去读取二进制数据就可以反序列化了。
另一种是不定长的浮动类型比如String或者基于String的JSON类型等。这种浮动类型的基础数据转化成二进制数组长度都是不一定的。对于这类数据通常的处理方式都是先往二进制数组中写入一个定长的数据的长度数据(Integer或者Long类型)然后再继续写入数据本身。这样反序列化时就可以先读取一个定长的长度再按照这个长度去读取对应长度的二进制数据这样就能读取到数据的完整二进制内容。 序列化机制是在高并发场景中非常重要的一个优化机制。高效的序列化实现能够极大的提升分布式系统的网络传输以及数据落盘的能力。例如对于一个User对象即可以使用JSON字符串这种简单粗暴的序列化方式也可以选择按照各个字段进行组合序列化的方式。但是显然后者的占用空间比较小序列化速度也会比较快。而Kafka在文件落盘时也设计了非常高效的数据序列化实现这也是Kafka高效运行的一大支撑。
在很多其他业务场景中也需要我们提供更高效的序列化实现。例如使用MapReduce框架时就需要自行定义数据的序列化方式。使用Netty框架进行网络调用时为了防止粘包也需要定制数据的序列化机制 消息分区路由机制
消息如何进行路由 Producer会根据消息的key选择Partition具体如何通过key找Partition呢 一个消费者组会共同消费一个Topic下的多个Partition中的同一套消息副本那Consumer节点是不是可以决定自己消费哪些Partition的消息呢
首先在Producer中可以指定一个Partitioner来对消息进行分配。
public static final String PARTITIONER_CLASS_CONFIG partitioner.class;
private static final String PARTITIONER_CLASS_DOC A class to use to determine which partition to be send to when produce the records. Available options are: ul liIf not set, the default partitioning logic is used. This strategy will try sticking to a partition until at least BATCH_SIZE_CONFIG bytes is produced to the partition. It works with the strategy: ul liIf no partition is specified but a key is present, choose a partition based on a hash of the key/li liIf no partition or key is present, choose the sticky partition that changes when at least BATCH_SIZE_CONFIG bytes are produced to the partition./li /ul /li licodeorg.apache.kafka.clients.producer.RoundRobinPartitioner/code: This partitioning strategy is that each record in a series of consecutive records will be sent to a different partition(no matter if the key is provided or not), until we run out of partitions and start over again. Note: Theres a known issue that will cause uneven distribution when new batch is created. Please check KAFKA-9965 for more detail. /li /ul pImplementing the codeorg.apache.kafka.clients.producer.Partitioner/code interface allows you to plug in a custom partitioner.;
这里就说明了Kafka是通过一个Partitioner接口的具体实现来决定一个消息如何根据Key分配到对应的Partition上的。可以很简单的实现一个自己的分配策略。
在之前的3.2.0版本Kafka提供了三种默认的Partitioner实现类RoundRobinPartitionerDefaultPartitioner和UniformStickyPartitioner。目前后面两种实现已经标记为过期被替换成了默认的实现机制。
对于生产者默认的Sticky策略在给一个生产者分配了一个分区后会尽可能一直使用这个分区。等待该分区的batch.size(默认16K)已满或者这个分区的消息已完成 linger.ms(默认0毫秒表示如果batch.size迟迟没有满后的等待时间)。RoundRobinPartitioner是在各个Partition中进行轮询发送这种方式没有考虑到消息大小以及各个Broker性能差异用得比较少。
另外可以自行指定一个Partitioner实现类定制分区逻辑。在Partitioner接口中核心要实现的就是partition方法。根据相关信息选择一个Partition。比如用key对partition的个数取模之类的。而Topic下的所有Partition信息都在cluster参数中。
//获取所有的Partition信息。
ListPartitionInfo partitions cluster.partitionsForTopic(topic);
然后在Consumer中可以指定一个PARTITION_ASSIGNMENT_STRATEGY分区分配策略决定如何在多个Consumer实例和多个Partitioner之间建立关联关系。
public static final String PARTITION_ASSIGNMENT_STRATEGY_CONFIG partition.assignment.strategy;
private static final String PARTITION_ASSIGNMENT_STRATEGY_DOC A list of class names or class types, ordered by preference, of supported partition assignment strategies that the client will use to distribute partition ownership amongst consumer instances when group management is used. Available options are: ul licodeorg.apache.kafka.clients.consumer.RangeAssignor/code: Assigns partitions on a per-topic basis./li licodeorg.apache.kafka.clients.consumer.RoundRobinAssignor/code: Assigns partitions to consumers in a round-robin fashion./li licodeorg.apache.kafka.clients.consumer.StickyAssignor/code: Guarantees an assignment that is maximally balanced while preserving as many existing partition assignments as possible./li licodeorg.apache.kafka.clients.consumer.CooperativeStickyAssignor/code: Follows the same StickyAssignor logic, but allows for cooperative rebalancing./li /ul pThe default assignor is [RangeAssignor, CooperativeStickyAssignor], which will use the RangeAssignor by default, but allows upgrading to the CooperativeStickyAssignor with just a single rolling bounce that removes the RangeAssignor from the list./p pImplementing the codeorg.apache.kafka.clients.consumer.ConsumerPartitionAssignor/code interface allows you to plug in a custom assignment strategy./p;
同样Kafka内置了一些实现方式在通常情况下也都是最优的选择。可以实现自己的分配策略。
从上面介绍可以看到Kafka默认提供了三种消费者的分区分配策略 range策略 比如一个Topic有10个Partiton(partition 0~9) 一个消费者组下有三个Consumer(consumer1~3)。Range策略就会将分区0~3分给一个Consumer4~6给一个Consumer7~9给一个Consumer。 round-robin策略轮询分配策略可以理解为在Consumer中一个一个轮流分配分区。比如0369分区给一个Consumer147分区给一个Consumer然后258给一个Consumer sticky策略粘性策略。这个策略有两个原则 在开始分区时尽量保持分区的分配均匀。比如按照Range策略分(这一步实际上是随机的)。 分区的分配尽可能的与上一次分配的保持一致。比如在range分区的情况下第三个Consumer的服务宕机了那么按照sticky策略就会保持consumer1和consumer2原有的分区分配情况。然后将consumer3分配的7~9分区尽量平均的分配到另外两个consumer上。这种粘性策略可以很好的保持Consumer的数据稳定性。
另外可以通过继承AbstractPartitionAssignor抽象类自定义消费者的订阅方式。
官方默认提供的生产者端的默认分区器以及消费者端的RangeAssignorCooperativeStickyAssignor分配策略在大部分场景下都是非常高效的算法。深入理解这些算法对于你深入理解MQ场景以及借此去横向对比理解其他的MQ产品都是非常有帮助的。
那么在哪些场景下我们可以自己来定义分区器呢例如如果在部署消费者时如果我们的服务器配置不一样就可以通过定制消费者分区器让性能更好的服务器上的消费者消费较多的消息而其他服务器上的消费者消费较少的消息这样就能更合理的运用上消费者端的服务器性能提升消费者的整体消费速度。 生产者消息缓存机制
Kafka生产者为了避免高并发请求对服务端造成过大压力每次发消息时并不是一条一条发往服务端而是增加了一个高速缓存将消息集中到缓存后批量进行发送。这种缓存机制也是高并发处理时非常常用的一种机制。
Kafka的消息缓存机制涉及到KafkaProducer中的两个关键组件 accumulator 和 sender
//1.记录累加器
int batchSize Math.max(1, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG));
this.accumulator new RecordAccumulator(logContext,batchSize,this.compressionType,lingerMs(config),retryBackoffMs,deliveryTimeoutMs, partitionerConfig,metrics,PRODUCER_METRIC_GROUP_NAME,time,apiVersions,transactionManager,new BufferPool(this.totalMemorySize, batchSize, metrics, time, PRODUCER_METRIC_GROUP_NAME));
//2. 数据发送线程
this.sender newSender(logContext, kafkaClient, this.metadata);
其中RecordAccumulator就是Kafka生产者的消息累加器。KafkaProducer要发送的消息都会在ReocrdAccumulator中缓存起来然后再分批发送给kafka broker。
在RecordAccumulator中会针对每一个Partition维护一个Deque双端队列这些Dequeue队列基本上是和Kafka服务端的Topic下的Partition对应的。每个Dequeue里会放入若干个ProducerBatch数据。KafkaProducer每次发送的消息都会根据key分配到对应的Deque队列中。然后每个消息都会保存在这些队列中的某一个ProducerBatch中。而消息分发的规则就是由上面的Partitioner组件完成的。 这里主要涉及到两个参数
//RecordAccumulator缓冲区大小
public static final String BUFFER_MEMORY_CONFIG buffer.memory;
private static final String BUFFER_MEMORY_DOC The total bytes of memory the producer can use to buffer records waiting to be sent to the server. If records are sent faster than they can be delivered to the server the producer will block for code MAX_BLOCK_MS_CONFIG /code after which it will throw an exception. p This setting should correspond roughly to the total memory the producer will use, but is not a hard bound since not all memory the producer uses is used for buffering. Some additional memory will be used for compression (if compression is enabled) as well as for maintaining in-flight requests.;//缓冲区每一个batch的大小
public static final String BATCH_SIZE_CONFIG batch.size;
private static final String BATCH_SIZE_DOC The producer will attempt to batch records together into fewer requests whenever multiple records are being sent to the same partition. This helps performance on both the client and the server. This configuration controls the default batch size in bytes. p No attempt will be made to batch records larger than this size. p Requests sent to brokers will contain multiple batches, one for each partition with data available to be sent. p A small batch size will make batching less common and may reduce throughput (a batch size of zero will disable batching entirely). A very large batch size may use memory a bit more wastefully as we will always allocate a buffer of the specified batch size in anticipation of additional records. p Note: This setting gives the upper bound of the batch size to be sent. If we have fewer than this many bytes accumulated for this partition, we will linger for the codelinger.ms/code time waiting for more records to show up. This codelinger.ms/code setting defaults to 0, which means well immediately send out a record even the accumulated batch size is under this codebatch.size/code setting.; 这里面也提到了几个其他的参数比如 MAX_BLOCK_MS_CONFIG 默认60秒 接下来sender就是KafkaProducer中用来发送消息的一个单独的线程。从这里可以看到每个KafkaProducer对象都对应一个sender线程。他会负责将RecordAccumulator中的消息发送给Kafka。 Sender也并不是一次就把RecordAccumulator中缓存的所有消息都发送出去而是每次只拿一部分消息。他只获取RecordAccumulator中缓存内容达到BATCH_SIZE_CONFIG大小的ProducerBatch消息。当然如果消息比较少ProducerBatch中的消息大小长期达不到BATCH_SIZE_CONFIG的话Sender也不会一直等待。最多等待LINGER_MS_CONFIG时长。然后就会将ProducerBatch中的消息读取出来。LINGER_MS_CONFIG默认值是0。
然后Sender对读取出来的消息会以Broker为key缓存到一个对应的队列当中。这些队列当中的消息就称为InflightRequest。接下来这些Inflight就会一一发往Kafka对应的Broker中直到收到Broker的响应才会从队列中移除。这些队列也并不会无限缓存最多缓存MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION(默认值为5)个请求。 生产者缓存机制的主要目的是将消息打包减少网络IO频率。所以在Sender的InflightRequest队列中消息也不是一条一条发送给Broker的而是一批消息一起往Broker发送。而这就意味着这一批消息是没有固定的先后顺序的。 其中涉及到的几个主要参数如下
public static final String LINGER_MS_CONFIG linger.ms;
private static final String LINGER_MS_DOC The producer groups together any records that arrive in between request transmissions into a single batched request. Normally this occurs only under load when records arrive faster than they can be sent out. However in some circumstances the client may want to reduce the number of requests even under moderate load. This setting accomplishes this by adding a small amount of artificial delaymdash;that is, rather than immediately sending out a record, the producer will wait for up to the given delay to allow other records to be sent so that the sends can be batched together. This can be thought of as analogous to Nagles algorithm in TCP. This setting gives the upper bound on the delay for batching: once we get code BATCH_SIZE_CONFIG /code worth of records for a partition it will be sent immediately regardless of this setting, however if we have fewer than this many bytes accumulated for this partition we will linger for the specified time waiting for more records to show up. This setting defaults to 0 (i.e. no delay). Setting code LINGER_MS_CONFIG 5/code, for example, would have the effect of reducing the number of requests sent but would add up to 5ms of latency to records sent in the absence of load.;public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION max.in.flight.requests.per.connection;
private static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_DOC The maximum number of unacknowledged requests the client will send on a single connection before blocking. Note that if this configuration is set to be greater than 1 and codeenable.idempotence/code is set to false, there is a risk of message reordering after a failed send due to retries (i.e., if retries are enabled); if retries are disabled or if codeenable.idempotence/code is set to true, ordering will be preserved. Additionally, enabling idempotence requires the value of this configuration to be less than or equal to MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE . If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled. ;
最后Sender会通过其中的一个Selector组件完成与Kafka的IO请求并接收Kafka的响应。
//org.apache.kafka.clients.producer.KafkaProducer#doSend
if (result.batchIsFull || result.newBatchCreated) {log.trace(Waking up the sender since topic {} partition {} is either full or getting a new batch, record.topic(), appendCallbacks.getPartition());this.sender.wakeup();}
Kafka的生产者缓存机制是Kafka面对海量消息时非常重要的优化机制。合理优化这些参数对于Kafka集群性能提升是非常重要的。比如如果你的消息体比较大那么应该考虑加大batch.size尽量提升batch的缓存效率。而如果Producer要发送的消息确实非常多那么就需要考虑加大total.memory参数尽量避免缓存不够造成的阻塞。如果发现生产者发送消息比较慢那么可以考虑提升max.in.flight.requests.per.connection参数这样能加大消息发送的吞吐量。 发送应答机制 生产者消息幂等性 生产者消息事务
