前言
Kafka 作为一个消息系统，其中很大的一个用途就是作为业务上的解耦，而它实现的模式就是经典的生产者消费者模式。毫无疑问，就出现了producer、consumer。然后消息总得有地方存放啊，然后就有了具体的broker，那在broker上是如何进行组织和存放的，就出现了partition。对应的为保证消息不丢失，也就出现了消息备份组这样一个概念（ISR，in-sync replica）再加上消息的topic也就形成了，kafka的 topic-partition-message 的三级负载结构。到这里Kafka中比较核心的几个概念就都有了，下面开始详细介绍。
producer
producer也就是生产者，是kafka中消息的产生方，产生消息并提交给kafka集群完成消息的持久化，这个过程中主要涉及ProducerRecord对象的构建、分区选择、元数据的填充、ProducerRecord对象的序列化、进入消息缓冲池、完成消息的发送、接受broker的响应。 具体的流程是这样的：
image.png
1、确定topic信息 2、确定value信息 3、然后进行消息的序列化处理 4、由分区选择器确定对应的分区信息 5、将消息写入消息缓冲区 6、完成消息请求的发送 7、完成消息响应的处理
ProducerRecord：
ProducerRecord 对象比较核心的信息有：topic、partition（这个信息是根据分区选择器来确定的）、key、value、timestamp
PS：时间戳信息是默认当前时间的，但是用户可以指定时间戳信息，但是不推荐这么做，broker中大体有这么几种log也就是消息存放文件普通日志文件，时间索引文件，普通索引文件。如果强行指定时间戳很有可能导致时间索引失效。
元数据：
元数据信息主要包括offset消息在分区日志中的位移信息、timestamp、topic/partition topic及对应的分区信息、checksum 消息对应的CRC32码、serializedKeySize 序列化后的key的字节数、serializedValueSize 序列化后的Value的字节数
Partition：
分区选择器，默认是murmur2 对于key进行hash计算然后对于总分区数求模以此得到被发送的分区号，当然我们实现producer时可以自定义partition，或者指定特定分区。
serializer：
serializer是kafka实现的自己的序列化工具用于将消息对象序列化成字节序列，Kafka中提供了ByteArraySerializer、ByteBufferSerializer、BytesSerializer、Long（Double Integer String）Serializer等几种序列化方法，用户也可以使用自定义的或者第三方的序列化工具。只需要使用指定对应参数即可（切记Kafka中指定对应的工具类时都是使用权限定名称来做的）
序列化相关的参数有如下：
key.serializer 针对各个部分做序列化方式
key.deserializer key.serializer对应解序列化方式
value.serializer对value部分指定的序列化方式
value.deserializer value.serializer对应解序列化方式
可以简单的理解为key要比value的应用范围广。
batch：
buffer.memory 指定producer待发送消息缓冲区的内存大小，默认32m，如果需要更改就使用这个参数进行修改。这里需要注意的是当producer端写消息的速度超过了专属IO线程发送消息的速度，并且缓冲区的消息数量超过buffer.memory指定的大小时，producer会抛出异常通知用户介入处理，这个缓冲区的大小需要根据实际场景来确定。
batch.size 指一个batch的大小，它直接决定了一个batch中存在的消息数量，这个直接与producer的吞吐量及延时等直接相关，因为所谓的micr-batch 是指原本应该串行一条条发送的消息更改为缓存一部分消息，等达到对应的消息规模时一次性发送，也不会像批处理规模那么大（主要为了平衡延时与性能，这个会有专门的篇章来介绍micr-batch）
linger.size
producer端会专门划出一部分内存用于待发送消息的缓存，batch.size决定了发送消息数量，同时间接决定了消息缓存时存在的延时。linger.size 就是针对这一点设计出来的，它决定了消息被投放进缓冲区时是否立马被发送，默认参数是0（立即发送），这个大多数情况下是合理的，但是会很大程度上拉低kafka的吞吐量。具体要根据实际的使用场景来确定了。
通信协议：
kafka 并没有使用现有的http协议等，而是在TCP 协议之上实现了自己的通信协议。单个client会创建多个socket链接与多个broker进行交互,Kafka 原生Java client使用类似于epoll的方式在单个连接上不停的轮训传数据，但是每个broker上只需要维护一个Scoket链接,保证了消息的请求的顺序处理，所以很清晰的可以看到在client端就需要我们自己去维护这个顺序了。
整体来说Kafka 中大约有三类连接：client与broker之间消息传输、controller 与所有broker之间的交互、client 获取元数据&rebalance的通信过程。
同其他协议类似，Kafka的通信协议的请求和响应也都是格式化的。由 固定长度初始类型（int8、int16、int32、int64）、可变长度类型（bytes、string）、数组。请求头由 api_key（int16，请求类型）、api_version（int16，请求版本号）、correlation_id（int32，与请求响应的关联号，这个字段就是给响应用的）、client_id（client id）
经常接触到的Kafka请求类型有：PRODUCE请求（生产消息请求）、FETCH请求（服务于消费消息，并不一定是clients向broker拉消息，也可能是follower副本向leader副本索要消息）、METADATA请求（获取指定topic的元数据信息：[topics] allow_auto_topic_creation）
PS：这里有一点需要说明，clients与broker是单向兼容的，这个在生产环境中如果不注意是格外容易发生问题的。这个兼容性是指，高版本broker可以兼容低版本clients，但是低版本broker无法兼容高版本clients，所以说升级clients版本，尤其是对接新的consumer时一定要格外注意。这个问题主要针对非Java client的，对于Java client来说，会自动判断连接的broker端所支持的client请求的最高版本。
producer interceptor
拦截器是新版本才出现的一个特性，并且是非必须的，interceptor 核心的函数有onSend（在消息序列化计算分区之前就被调用）、onAcknowleagement（被应答前或者说在发送失败时，这个方法是运行在producer的I/O线程中的，所以说如果存在很多重逻辑的话会导致严重影响处理消息的速率）、close。通常是通过为clients定制一部分通用且简单的逻辑时才会使用的。
压缩算法
Kafka支持的压缩算法还是很可观的：GZIP、Snappy、LZ4，默认情况下不进行消息压缩，毕竟会消耗很大一部分cpu时间，导致send方法处理时间变慢。启动LZ4 进行消息压缩的producer的吞吐量是最高的。
