Kafka源码

checkstyle

bin 执行脚本

clients 客户端(java写的)

connect 获取数据源

core 消息系统(Scala写的)

examples 使用样例

gradle

streams 流式计算和处理(java)

代码结构

在一个JVM内部，如果要搞多个kafkaproducer的话，每个都会默认生成一个client-id，producer-自增长数字

Partitioner 分区器：用来决定发送的每条消息路由到Topic的哪个分区

启动的时候会拉取一次，后面每隔一小段时间会再次发送请求刷新元数据 metadata.max.age.ms 默认值5分钟

发送消息的时候如果发现某个topic本地没有metadata，会尝试拉取这个topic的元数据

Metadata

maxRequestSize 请求最大大小 默认1MB

totalMemorySize 缓冲区大小 默认32MB

retryBackOffMs 重试时间间隔 默认100MS

maxBlockTimeMs 缓冲区填满后阻塞时间 默认60S

requestTimeoutMS 请求超时时间 默认30s

核心参数

batchSize 一次请求消息打包的大小 默认16K

lingerMS 设置指定时间，如果batchSize没凑够也会发送消息

RecordAccumulator 缓冲区 负责消息复杂的缓冲机制

MaxInFlightRequestsPerConnection 同一个broke最多能受到几个没有响应的请求 默认5

ConnectionMaxIDLE空闲网络连接的最大数量

一个网络连接最多空闲时间 默认9分钟

重试连接的时间间隔 默认5MS

SendBuffer Socket发送缓冲区的大小 默认128K

ReceiveBuffer Socket接收缓冲区大小 默认32K

网络通信组件 NetworkClient

acks 默认是1  只要leader写入成功就返回

retries 失败重试次数 默认0

Sender线程类叫“KafkaThread”，线程名字叫\"kafka-producer-network-thread\" 负责从缓冲区获取消息发送到broke上去

序列化组件

拦截器组件

KafkaProducer核心组件

并没有真正的去某一个broke上拉取元数据，但是把配置的broke地址转化为node，放在cluster对象实例里，对集群元数据做了初始化，在后面根据发送消息时候的topic，再去拉取

KafkaProducer在初始化的时候会不会从集群拉取元数据？

回调自定义的拦截器

序列化key和value，将数据转换成字节数组

没有指定key时，kafka会使用自己维护的一个AutomicInteger类型的 自增counter均匀的把消息分配到这个topic的分区上

如果指定了key，kafka会使用自己实现的murmur2工具类将key这个字节数组转换为一个hash值，然后对分区数进行取模，可以保证相同的key都在一个分区

分区策略

基于获取到的topic元数据，使用Prititioner组件获取消息对应的分区

检查要发送的消息是否超出消息最大大小，和内存缓冲最大大小 

BufferPool里面有一个deque作为队列缓存一些ByteBuffer内存空间，可以复用，大小是batch大小

BytePool数量 * batch 大小 = 已经缓存内存空间大小

availableMemory是剩余还可以使用的空间大小

将消息添加到内存缓冲里去

设置好自定义的callback回调函数以及对应拦截器的回调函数

batch已满

时间超过linger.ms

内存耗尽

客户端要关闭了

强制flush所有数据时

bath是如何判定可以被发送出去的

如果不满足上面条件，如何建立连接

判断目标broke是否可以发送请求过去 

如果某个分区对应的batch满了或者创建了新的batch，此时就会唤醒sender线程发送batch

KafkaProducer.send()执行流程

KafkaProducer

解耦

可恢复性

缓冲

灵活性&峰值处理能力

异步通信

Kafka是一个分布式的基于发布/订阅模式的消息队列，主要应用于大数据实时处理领域

点对点

发布/订阅

模式

消息队列

概述

Kafka源码