Kafka源码简要分析-Toy模板网

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

一、生产者的初始化流程

二、生产者到缓冲队列的流程

三、Sender拉取数据到Kafka流程

四、消费者初始化

五、主题订阅原理

六、消费者抓取数据原理

七、消费者组初始化

八、消费者组消费流程

九、提交offset原理

一、生产者的初始化流程

首先获取事务id和客户端id（用到事物必须要事物id不然报错，每个生产者都需要唯一标识客户端id）
监控kafka相关情况的JmxReporter配置
然后获取分区器，如果用户有自定义的就读取配置的，如果没有配置就用默认分区器
然后key和value进行序列化
然后就读取自定义拦截器，可以定义多个拦截器，组成拦截器链
然后初始化控制单条日志的大小，默认是1m；缓冲区大小，默认32m；
创建内存池，缓存队列，初始化批次大小默认16k，压缩相关处理，默认是none，重试间隔时间默认100ms
连接kafka集群，获取元数据，才能知道要发送到哪个分区
创建sender线程，会有个创建sender的方法，sender线程负责拉取缓冲队列消息到Kafka，在方法里面会定义缓存请求的个数默认5个，然后请求超时的时间，然后创建一个网络请求客户端对象，会传入刚刚的参数还有客户端id，重试时间，发送缓冲区的大小128和接受缓冲区的大小32，还有acks等配置。sender继承了Runnbale接口，然后会new个sender线程出来用上面这些参数，然后返回。
sender放到后台，启动sender线程

在执行到拦截器的时候就要调用一个onSend方法，如果有多个拦截器，每个拦截器都会走一次这个方法，这个方法就是拦截器对数据加工的
然后获取元数据，要根据主题的分区放到对应的缓存队列
序列化相关操作key和value的序列化和压缩
分区操作，如果指定了分区，直接分配到指定分区；没有指定就会根据分区器进行分配，没有指定key就会粘性分区处理（如果批次大小和活着时间到了不然就一直是那个，满足才能创建新队列用），如果指定key就根据key到hashcode进分区数取模，
保证（序列化和压缩后）数据大小能够传输，他去读取配置的消息最大值和缓冲区大小，如果有超过的抛异常
向缓存队列里面追加数据，获取或者创建一个队列按照分区，然后尝试添加数据（一般不成功，因为还没申请内存），然后根据16k和现在压缩后的总大小取最大值，申请内存就申请这个大小，内存池分配内存，然后sender线程拿走就了会释放内存。
如果批次大小满了或者有了新的批次需要创建，就唤醒sender线程把缓冲队列的数据拉取过去。