读kafka生产端源码，窥kafka设计之道（上）-Toy模板网

这篇具有很好参考价值的文章主要介绍了读kafka生产端源码，窥kafka设计之道（上）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1. kafka 高吞吐之道-------异步提交批量发送

简约的发送接口----后面隐藏着并不简单的设计

kafka发送消息的接口非常简约，在简约的表面上，其背后却并不简单。先看下发送接口

kafkaProducer.send(new ProducerRecord(topic,msg), new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (null != exception) { //消息发送失败，记个日志
                    log.error("send to kafka error {}",exception);
                }
            }
        });

正常情况下，调用kafkaProducer.send方法，由业务线程执行（比如tomcat的业务线程）；业务线程在执行send代码后，就直接返回；这个send方法的本质：其实是把消息写入到了本地内存中，并没有发生网络IO。

什么时候会把内存中缓存的消息发给broker集群了？
是kafka的Sender Thread来负责发送的（代码类名就叫 Sender）；Sender 线程会把内存中的数据，重新组织：把属于同一个broker的多条消息，打包成一个ProduceRequest；通过Sender Threader，把多条数据一次网络IO批量发给到对应的broker。

整个过程如下图：
读kafka生产端源码，窥kafka设计之道（上）,kafka,kafka,分布式,架构,中间件

图1 消息的生成和发发送过程

从图中可看到：业务线程投递消息到内存是同步的。但这一过程速度非常快。到底有多快？可以借鉴多年前Google的Jeff Dean 在Engineer All-Hands Meeting上对I/O延迟的测试数据
读kafka生产端源码，窥kafka设计之道（上）,kafka,kafka,分布式,架构,中间件

图2 IO延迟

写内存延迟在100纳秒；写网络延迟在20,000纳秒；两种延迟相差200倍；kafka提供的send接口，翻译过来是发送的意思；这个表面的“发送”接口，理论上调用百W次，只花费了1秒钟。

为什么kafka发消息可以使用“异步批量”方式了？而没有选择“同步单条”方式了？

推测有以下2个主要原因

原因1：异步，消息中间件自带的好处之一
使用消息中间件的好处之一是异步（另外两个是解耦，削峰）。即发送消息和消费消息不是一个同步的过程；很有可能生产者发送消息成功了，而消费者消费消息失败了；并且生产消息和消费消息可以是在不同的时间点和不同机器上发生的。从消息产生和消费的整体流程上来看，不需要同步。 kafka生产端能够使用这种方式，和他的应用场景是密切相关的，即天生就是异步的。

原因2：通过减少网络次数和传输的元数据方式，整体提高了发送端的吞吐量

假设生产端产生一条消息，立即同步发送给服务端；相比较攒一波再发送；必定会增加网络I/O次数，
加重了网络传输的资源消耗。

这让我想到了亲身项目经历：当时有一个IOT项目；需要在30秒内，完成10W设备的连接，并且在连接完后，需要把设备版本号进行入库操作。当时服务端采用的是连接一个设备，存一次当前设备版本到库里；这种方式在测试环境进行压测情况下，应用响应奇慢无比和宕机没啥区别了，mysql数据库的cpu比平常高了70%； 后来采用这种异步批量提交版本到数据库的方式；应用轻松应对10W个设备同时发起连接，并保存版本号到数据库；而mysql的cpu也只比平常高了1%；

攒一波的本质，带来的最终效果是一次IO发送多条消息；而且这种批量打包消息还不是针对某个topic 分区的，也不是针对某个topic的，而是针对broker维度的，即kafka生产端的IO线程，会在发送消息的时候，会重新计算每条消息对应的是那个broker，然后把发往同一个broker的消息在打包成一个ProduceRequest；带来的最终好处：把多个topic，多个分区的数据打包发往一个broker时只需一次网络I/O。

2. 无锁化的线程模型----性能提高的利器

从前面的图1，我们发现；在一个应用中，其实只用了一个IO单线程，对集群所有的broker进行网络IO通信；而采用IO单线程的方式。有以下优点：

简化编程模型

使用单线程管理网络 I/O 可以简化编程模型，避免了多线程编程中的并发和同步问题。开发人员可以专注于业务逻辑的实现，而不必过多关注线程间的同步和数据一致性问题。

降低资源消耗，提高性能

在传统的BIO网络编程模型中，通常用每个连接一个线程或进程的方式进行处理；见下图：

读kafka生产端源码，窥kafka设计之道（上）,kafka,kafka,分布式,架构,中间件

图3 传统BIO模型

在这种方式里，线程数和网络连接数成线性关系。 而线程是一种资源；过多的线程会导致占用更多的内存并且管理线程本身也会有资源开销，过多线程产生的上下文切换，也会导致系统性能直线下降。

kafka采用单线程异步非阻塞的方式管理网络 I/O，自身通过事件循环的机制，高效地处理并发请求，避免了线程切换和阻塞操作的开销。

想想在java界鼎鼎大名的网络编程框架netty，不就是这种无锁化的IO线程设计吗？ 1个线程管理多个TCP连接的读写操作。

保证消息发送的顺序性

Kafka 是一个分布式消息系统，消息的顺序性对于一些业务场景非常重要（比如订单状态的变化，需要通知到下游进行对应的业务处理）。使用单线程管理网络 I/O 可以保证消息的发送顺序，因为所有的发送操作都在同一个线程中进行，不会出现并发发送导致消息顺序混乱的情况。 注：kafka的消息顺序，只保证在broker维度是顺序的，不保证消息在多个broker间是顺序的，这和kafka是个分布式多分区特性有关。

3. 良好的封装 -----开箱即用，减少使用者心智

1、生产端只需要7行代码，就可接入kafka，并且具备消息发送能力

Map<String,Object> produceConfigMap = new HashMap<String,Object>();
produceConfigMap.put("bootstrap.servers",KafkaConfig.getInstance().getProperty("bootstrap.servers"));
produceConfigMap.put("retries",3);
produceConfigMap.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
produceConfigMap.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
produceConfigMap.put("acks","all");
KafkaProducer kafkaProducer = new KafkaProducer(produceConfigMap);

2、IO线程封装：生产者端和kafka broker集群所有的网络交互，无需上层关心，即和中间件进行网络交互的业务，交给sdk 的IO线程就OK了

3、元数据的更新，由sdk内部完成，对客户端的外部使用者并不可见。
不仅做到了元数据的自动更新；还做到了元数据更新后，客户端的重连，消息的重试，让
生产端消息尽量不丢; leader分区迁移或者leader分区切换时，客户端自动感知和切换（这个高级功能是kafka整体高可用的前提）；前段时间在做leader分区迁移和切换时；生产端是秒级感应；在leader分区切换后，生产端通过自动重试的机制把消息发给了新分区，保证了消息的不丢和kafka的生产端高可用

4. 扩展性考虑

生产端扩展性考虑，主要有两点：

第一点：通过拦截器，在消息发送之前，统一的做点什么事；是不是感觉到有点熟悉的面向“AOP”编程味
比如：在实现ProducerInterceptor接口，在onSend方法里；对所有消息发送记录以打印日志的方式进行单独记录

第二点：从消息对象到网络传输的字节，即序列化方式，和反序列化方式；可自行定义实现；
比如你觉得原生的字符串方式的序列化不好，序列化后的字节数太大，加重了网络传输的负载；你可以使用业界优秀的protobuf，hessian，kryo等二进制的序列化方式；只需实现序列化接口，反序列化接口

5. 其它好玩的事情

在看kafka生产端源码时，除了发现有很多设计上的优点以外，还发现其在使用底层技术上也很有考究：
比如：

不走寻常路的生产者消费者模型

kafka消息生成和发送，其本质是一个生产者消费者模型；而生产者消费者模型，我们一般用什么做队列了？首先想到的是并发包里的BlockingQueue做队列；
为啥了？
第一：BlockingQueue底层帮我们解决了并发访问的问题；
第二：绝大多数现有讲解生产者消费者模型的课程都会用此队列，已经形成了使用此模型的第一联想位；

但kafka出乎个人意料之外的使用了双端队列；why?