一种物联网设施管理系统中的Kafka优化方法∗

顾 鑫 程淑玲

（1.武汉邮电科学研究院 武汉 430074）

（2.烽火通信科技股份有限公司 武汉 430074）

1 引言

近些年，随着物联网、大数据、云计算等技术的快速发展，需要分析、处理和传输的信息量飞速增长［1］。将物联网技术应用于人井、光交箱、室外柜、电子锁等物联设施可以大大地提高对这些设施的管理效率［2］。我们的基础设施物联管理系统就是在这样的背景下建立的一个平台，旨在借助于这些技术对物联设施的运行维护提供更准确的实时检测、指标分析及其他的一些服务，保障物联设施安全、稳定、高效地运行。

本系统作为一个管理十万级物联网设施的大数据平台，消息中间件起着关键的作用［3］。设施上报的原始数据不直接进入处理程序，而是先存入Kafka消息队列中。基于消息队列，独立的节点无需紧耦合工作，能够以异步的方式交换信息，从而可以提高系统的灵活性［4］。虽然消息中间件的应用能带来收益，但它造成的负面影响也是需要我们关注的。在大规模的分布式系统中，消息中间件的表现已经成为制约系统整体性能的重要瓶颈［5］，为了更高效地利用Kafka消息中间件，降低其造成的延时和资源消耗是十分有必要的。

2 选用Kafka的原因

2.1 Kafka介绍

Kafka是一个分布式的、基于发布/订阅的开源消息系统，广泛地用于各种消息交换系统、实时数据流应用和大数据平台中［6］。它的数据组织是按主题来划分的，每个主题就是一个消息流［7］。主题又被分解成多个分区，同时也被复制多份。消息生产者向指定的主题中发布消息，消息消费者便可以获取消息［8］。一个Kafka集群由一个或多个broker节点组成。

2.2 Kafka的优势

Kafka与传统的消息系统相比有着很多优势之处：被设计成分布式的消息集群［9］，适合于演进式系统，随着业务量增加可以动态伸缩；消息可以被持久化存储，这样除了支持实时消息处理外，也能做到离线批量数据分析处理；消息发布和消息订阅的吞吐量非常巨大，性能优异［10］；当集群中有节点发生故障时，能自动重新负载均衡，容错性强；多个微服务之间的消息交互也可以使用Kafka来交换，这种响应式的异步数据交换模式能提高系统的性能。

3 Kafka造成的延时和资源消耗

系统的数据流如图1所示。物联设施的告警信息可能存在告警风暴的场景，为了降低告警风暴对系统造成的冲击，添加Kafka消息中间件是十分有必要的。Kafka将接收到的告警信息存放在消息队列中，异步地将这些信息以数据流的形式传输到数据分析层和数据存储层［11］。

图1 系统的数据流

3.1 延时

通常情况下，Kafka消息中间件的缓存越大，造成的额外延时就会越高［12～13］。由于系统对实时性的要求较高，需要找出合适的缓存大小来确保低延时。我们以每条0.2KB，每秒10000条的速度发送JSON格式的消息，测量了Kafka消息中间件在不同缓存大小造成的额外延时，如图2所示。经比较发现，添加16MB缓存的三节点Kafka集群几乎不会额外地增加系统延时，和不添加Kafka消息中间件一样，都是有99.95%的消息低于100ms延时的门限。所以，在物联设施管理系统中使用16MB以下缓存的Kafka集群是可行的。

图2 不同缓存大小的Kafka延时

3.2 额外CPU资源消耗

由于16MB以下缓存的Kafka消息中间件几乎不会增加系统延时，我们测量了添加16MB以下不同缓存大小的Kafka集群造成的额外CPU资源消耗，如图3所示。缓存大小在1MB时，添加Kafka消息中间件造成的额外CPU资源消耗最低，为5.4%。

图3 不同缓存大小的Kafka额外CPU资源消耗

4 优化方案

Kafka消息中间件带来的额外CPU资源消耗有比较大一部分原因是KafkaProducer的Serializer接口不必要的内存分配和复制导致的［14］，由于低效的内存管理导致的垃圾回收暂停会增加CPU的使用率［15］。Kafka库中的Serializer接口如图4所示。可以注意到，序列化方法返回byte［］字节数组，这实际上迫使计算机创建一个总是分配新字节数组内存空间和复制内存内容的实现。由于这种API设计，KafkaProducer在堆内存上分配了过多的字节数组，不能够重用已分配的内存，特别是当对象序列化为可变长度数组的时候。

图4 Kafka的Serializer接口

java.nio.ByteBuffer充当连续内存的可变指针，能够记录开始和结束的位置。在它的帮助下我们可以预先分配并重用内存，以避免为每个序列化对象都分配新的内存空间。为了在优化时对现有代码产生的影响尽可能小，在Serializer接口中添加了一个带有default实现的方法，如图5所示。在Seri⁃alizer实现中重新声明serializeAsByteBuffer方法，在此更改之后，可以直接使用ByteBuffers来替换字节数组，返回指向序列化对象开始和结束的预分配内存。优化之后，我们重新测量了不同缓存大小的三节点Kafka集群造成的额外CPU资源消耗，与优化之前的对比结果如表1所示。可以看到，优化后缓存大小在2MB时额外的CPU资源消耗最低，为3.3%。

图5 添加到Serializer接口中的方法

表1 优化前后Kafka不同缓存大小的额外CPU资源消耗

5 优化的意义

我们的物联设施管理系统是部署在云端的，计算资源的利用成为了关键的影响因素之一，因此如何降低资源消耗是值得研究的。从优化前使用1MB缓存的Kafka消息中间件的5.4%最低额外CPU资源消耗到优化后使用2MB缓存的Kafka消息中间件的3.3%最低额外CPU资源消耗，此优化方案让Kafka造成的额外CPU资源消耗降低了39%。由于系统管理的物联设施数量巨大，未来还会有新的设施接入，伴随着数据量的增加，Kafka集群会根据需要进行扩展，这样一来优化带来的收益会进一步增加。

6 结语

本文针对KafkaProducer的Serializer接口提出了优化方案，显著地降低了使用Kafka消息中间件带来的额外CPU资源消耗，提高了物联设施管理系统的性能，同时也为将来伴随系统扩展而扩展的Kafka集群的使用打下了基础。除了Serializer接口，还有其他的方面可以对Kafka消息中间件进行优化，这也是未来需要继续探究的。另外，这种优化消息中间件的思想也可以应用在Kafka以外的其他消息中间件和消息系统。