Kakfa系列_消息发送流程
一、重要配置项 1、bootstrap.servers//Kafka服务器连接地址,必填 2、key.serializer//用于消息的key做序列化,必填 3、value.serializer//用于消息的value做序列化,必填 4、acks acks=0,表示producer只负责发送,不关心服务端处理结果。可以以理解为以oneWay的方式发送 acks=all或者-1,表示producer会等到服务端topic副本都写入个字的本地日之后,才会把发送结果返回给producer acks=1,表示producer发送后,topic副本的主节点写入成功后就把发送结果返回给producer,不关系其他副本写入情况 5、buffer.memory//producer用于缓存消息的缓冲区大小,默认32MB。 6、compression.type//producer消息压缩策略,默认关闭。Kafka支持3种压缩策略:GZIP、Snappy、LZ4 7、retries//消息发送失败,producer内部重试次数 8、batch.size//重要。默认16KB,producer默认会将同一分区下的消息封装到一个batch中 9、linger.ms//默认值0,表示消息需要立马发送到服务端。该配置可与batch.size配置一起结合使用进行调优 10、max.request.size//表示单条消息大小 11、request.timeout.ms//producer发送到Kafka服务端的超时时间,默认30s 二、发送流程 1、确认数据要发送到的 topic 的 metadata 是可用的(如果该 partition 的 leader 存在则是可用的,如果开启权限时,client 有相应的权限),如果没有 topic 的 metadata 信息,就需要获取相应的metadata; 2、序列化 record 的 key 和 value; 3、获取该 record 要发送到的 partition(可以指定,也可以根据算法计算); 向 accumulator 中追加 record 数据,数据会先进行缓存; 如果追加完数据后,对应的 RecordBatch 已经达到了 batch.size 的大小(或者batch 的剩余空间不足以添加下一条 Record),则唤醒 sender 线程发送数据。详见流程图:
|