【Kafka实战】智慧工厂IOT项目Kafka消息积压优化方案

概述

Kafka 系统在本公司的智慧工厂 IOT 项目中得到了广泛应用。目前，全国包括海外有上千个矿山工厂、光伏工厂、Cableway工厂等使用我们的系统，虽然这些工厂的业务层逻辑各不相同，并且对应的服务系统也不完全一样。但是作为消息中间件和缓存中间件的Kafka集群和MQTT集群整个公司都是使用的同一套底层服务，所以也面临着巨大的负载压力。随着业务的迅猛发展， Kafka 集群规模快速增长。目前，公司的 Kafka 集群每日处理消息总量已经达到达到数十亿级别，峰值消息处理速度在千万条/s，随着业务规模的扩大，我们也面临着新的问题和技术挑战也逐渐增加。

业务场景

.png)

从业务场景上来说，Kafka集群在本项目中主要被分为了3类集群，分别是消息集群、Log集群和持久化集群，以满足不同业务的不同需求。

但是对于IOT项目来说，我们的项目与互联网项目有所不同，我们还有一个由MQTT集群构成的IOT数据采集集群，之所以不采用Kafka的原因主要是针对物联网场景下网络信号不稳定和数据包大小限制的原因。

此外，作为整体解决方案，我们还引入了数据Mirror服务（Web Service），专门负责将数据从消息集群、LOG集群传输、IOT数据集群传送到持久化集群以及分发到web前端进行实时展示。

因此，本智慧工厂IOT项目的集群构成如下：

1. 消息集群（Kafka）：Kafka作为消息中间件，为不同的在线业务提供异步的消息通知服务。

2. LOG 集群（Kafka）：业务程序将生成的日志直接发送至Kafka，通过Kafka进行高效的传输与收集。由于数据不落地，LOG集群对Kafka的可用性要求极高。这个集群还为重要的实时计算和模型训练提供了稳定而高效的数据源。

3. 持久化集群（Kafka）：LOG数据的最终汇聚点，实时将数据转储到MongDB集群当中，用于离线处理。除了提供持久化数据，持久化集群还为次要的实时计算和实时训练提供了关键的数据支持。

4. IOT数据集群（MQTT）：该集群的消息来源比较复杂，有工厂现场的嵌入式设备直接发送的IOT数据包，也有工厂现场服务处理之后发送的数据包。

对Kafka进行物理集群划分的主要目的在于确保服务质量，以及限制Kafka集群问题对整体系统的影响。这种集群划分的设计有助于更好地掌控和维护Kafka在不同业务场景中的性能和可用性。

业务规模

如上表所示，我们拥有一个相对庞大的 Kafka 集群，具体参数如下：

• 总机器数量： 200台，为集群提供了充足的计算资源。

• 集群数量： 8个独立的集群，每个集群在处理特定任务和数据流时有一定的独立性。

• Topic数量： 系统中定义了1200个主题，每个主题代表着一类特定的数据流或信息分类。

• TP（Topic Partition）数量： 2万个分区，分布在整个集群中以实现更有效的数据处理和负载均衡。

• 每日处理消息数： 集群每天处理着数十亿条消息，这些消息可能来自不同的业务流程和数据源。

• 峰值消息处理能力： 具有百万条每秒的消息处理峰值，显示了其在处理高峰时期的出色性能。

这个规模的 Kafka 集群为百亿级别的每日消息处理和百万消息处理能力的峰值情况下，提供了强大的支持，确保了数据的高效管理和分发。

Kafka集群消息积压优化

随着用户规模不断扩大，目前全国包括海外有上千个矿山工厂、光伏工厂、Cableway工厂等使用我们的系统，虽然这些工厂的业务端逻辑各不相同，并且对应的服务系统也不完全一样。但是作为消息中间件和缓存中间件的Kafka集群和MQTT集群整个公司都是使用的同一套底层服务。工厂规模扩大的同时，随之而来的是消息量越来越来多，导致时常发生消费者处理不过来，消息积压的情况时有发生。

由于工厂IOT设备时时刻刻都在发送IOT数据到Kafka集群，而Kafka集群负责将数据持久化到MongoDB中，在数据的峰值时刻（IOT项目的数据峰值一般都比较稳定），可以达到百万条/s。这些数据都需要按照时间顺序被持久化到MongoDB集群中，给算法团队进行分析以及进行工业大数据预测。

当消息积压的情况发生的时候，由于Kafka Broker的磁盘空间是非常有限的，而大量消息出现堆积的情况时可能会造成日志被删除，导致消息丢失的情况发生。

虽然说对于消息积压的问题，直接扩容增加TP数目是一个很好的解决方案，但是公司大量采购设备的审批流程复杂，并且本着省钱的惯例，先做系统优化。以下是一些在生产实践过程中所遇到的关于Kafka消息积压的相关问题和解决方案，这里做一个简单且汇总。

IOT数据包过大

问题原因：

如上图所示，尽管Kafka宣称支持百万级的TPS，然而，在消息从生产到消费的过程中，涉及到多次网络IO和磁盘IO操作，对于消息体过大的情况，这些IO操作的耗时会进一步增加，直接影响了Kafka的生产和消费速度。此外，慢速的消费者会导致消息积压，而消息体过大还可能浪费服务器磁盘空间，甚至引发磁盘空间不足问题。

在面对这些问题时，需要有针对性地优化消息体过大的情况，总的来说，当数据包大小过大的时候会有以下问题：

1. IO操作导致的性能影响： Kafka的生产者发送消息到Broker，Broker写数据到磁盘，以及消费者从Broker获取消息，这三个环节分别涉及一次网络IO和一次磁盘IO（写或读操作）。对于一次简单消息的完整过程，共需经过2次网络IO和2次磁盘IO，增大消息体将进一步加剧IO操作的耗时，对整体性能产生负面影响。

2. 消费者速度不足导致积压： 慢速的消费者会导致消息积压，使系统处理消息的能力受到限制，从而影响实时性和性能。

3. 磁盘空间浪费及不足问题： 大消息体会占用更多的磁盘空间，可能导致服务器磁盘空间的浪费。若不加以注意，可能会引发磁盘空间不足的问题，影响系统的正常运行。

优化策略：以Cableway工厂为例

.png)

• IOT数据消息拆分：

  每条索道上存在着上百个吊厢，每个吊厢有其单独的IOT设备来实时获取吊厢的定位信息，监控信息等数据。

  在项目的初始阶段，由于数据量不大，是将每条索道的所有吊厢数据经现场服务采集处理之后再组合成一个Json数据包发送给云端的MQTT集群的，数据再从MQTT集群按照原格式发送给Kafka集群进行持久化和消息分发。

而现场的监控数据信息会经过工厂现场部署的微服务进行处理和本地持久化之后，再按照时间间隔采样之后传送到Kafka服务器最后在云端进行持久化操作。

因此，我们将这些过大的IOT数据包进行了拆分，拆分成若干个小的IOT数据包再发送到MQTT集群，最后转发到Kafka集群。

• IOT数据消息压缩：

  IOT设备本身提供的是串口服务器编码的字节流数据，项目最初为了满足现场数据的实时展示，会在本地对IOT的字节流数据进行解码再转发到MQTT服务器，这一过程IOT数据的编码格式由字节流编码变成了Json格式的数据，大大增加了传送数据包的大小。

  因此，优化方案中，将字节流IOT数据直接丢给MQTT集群并进行转发，由持久化Kafka集群拿到数据之后，再改集群的处理流中进行数据的解码和持久化操作。

通过以上2个消息包优化策略，在一定程度上解决了消息积压的问题。

消费者线程模型问题

接着上面的问题来说，我们在生产者端将数据进行压缩处理和缩小包大小之后，Producer端确实磁盘的IO压力降低了，生产者的消息推送速度也随之变快了，同时，消费者端的磁盘IO压力也变小了，接收消息的速度也更快了，理论上来说，整个系统的消息积压问题会得到一定程度上的缓解。

但是，通过Kafka Manager监控发现，此时整个系统的消息积压程度并没有得到改善，反而更加严重了。分析消费者的日志我们发现，虽然IOT数据包大小变小了，但是我们其实是将本来由生产者执行的消息解析的逻辑丢给了消费者端来执行。这样一来，消费者端在进行数据解析的过程中耗费了大量的时间，是的消费的速度相对于以前降低了很多，这就使得消息积压反而更严重了。

于是，我们团队思考通过优化消费者端线程模型的方式来增加消费者消费消息的速度，缓解消息积压的压力。

Kafka的消息拉取模型

在介绍我们的线程优化方案之前，先了解一下Kafka消费者的消息拉取模型。

如上图所示，Kafka是采用poll模式对Broker中的消息进行拉取的：

• 当客户端请求消息的时候Kafka Consumer会先尝试从本地缓存获取，如果获取到了消息则直接返回。
• 如果本地缓存中没有消息，则真正调用网络请求从Kafka集群中拉取，并且直接返回。
• 当网络请求结束，Kafka Consumer通过回调的方式来获取数据。

但是，由于出于消费者性能上的考虑，Kafka Consumer的线程模式是非线程安全的，用户无法在多个线程中共享一个Kafka Consumer实例，当Kafka Consumer检测到多线程访问的时候会直接抛出异常。

原本的消费者线程模型 —— 每个线程维护一个 Kafka Consumer

项目中原本的Kafka的消费模型虽然也是采用的线程池的方式，但是采用的是每一个线程维护一个Kafka Consumer的方式来实现的，如下图所示：

.png)

这种消费模式下，每个线程管理一个 Kafka Consumer，从而实现线程隔离消费。由于每个分区在同一时刻只能由一个消费者处理，因此这种模型天然支持顺序消费。但是这种模型的缺点在于无法有效提升单个分区的消费能力。

在我们的项目中一个主题拥有大量分区，只能通过增加 Kafka Consumer 实例的方式来提高整体消费能力，但是这也导致了线程数量的增加，也会产生了项目 Socket 连接的巨大开销。

优化后的线程模型——单Kafka Consumer实例 + 多 worker 线程

.png)

这种消费模型将 Kafka Consumer 实例与消息消费逻辑解耦，从而实现多线程消费而无需创建多个 Kafka Consumer 实例。它还具备动态调整 worker 线程的能力，使其根据消费负载情况进行灵活调整。这种模型在提高并发性能方面表现出色，却无法保证消息的顺序性。

如果需要确保消息按照特定顺序被消费，就需要在采用第二种消费模型的基础上进行修改，这里我们采用了如下方案来解决消息的顺序消费问题：

1. 消费线程池初始化：

在初始化阶段，对消费线程池进行初始化。创建若干个单线程线程池，数量由 threadsNumMax 决定。每个线程池的目的是为了保证每个分区取模后能获取到一个线程池，以实现串行消费。

2. 消息分配到线程池：

对消息分区进行取模，根据取模后的序号从线程池列表缓存中获取一个线程池。相同分区的消息分配到相同线程池中执行，确保相同分区的消息串行执行，实现消息的顺序性。

3. 位移提交机制：

在消费前，创建一个 CountDownLatch，计数为本次消息数量。消费逻辑中，在 finally 中执行 countDown 操作，主线程阻塞等待本次消息消费完成。为了保证手动提交位移的正确性，只有当本次拉取的消息全部消费完毕后才进行位移提交。

4. 防止消息积压：

如果某些分区的消息堆积量少于500条，会继续从其他分区拉取消息，以达到每次拉取500条消息的目标。当每个分区的积压超过500条时，每次拉取的消息中很可能只包含一个分区。在这种情况下，利用分区取模将同一个分区的消息放到指定的线程池中进行消费，确保消息的顺序性。

以上机制保证了在各种情况下，能够有效实现顺序消费线程模型。

MongDB单集合数据量过大

当MongoDB单表数据量过大的时候会影响数据插入效率，即使采用了sharding模式，插入性能也会收到很大的影响，从而影响我们Kafka整个集群的消费效率。为此，本文先对MongoDB单表插入性能受集合大小的影响程度进行展示，然后分析解决办法。因为我们主要讨论插入效率导致的Kafka消息阻塞，因此这里只列出MongDB插入性能的测试数据。

MongoDB单集合插入测试

如上图所示，为测试MongoDB集群的插入性能开销示意图，测试服务器使用的是CentOS操作系统，24GB内存。这里分别测试了MongDB的单条记录插入和批量插入在单机模式和Sharding模式下的插入效率。可以看出，当MongDB单集合数据量到达2000万条的时候，无论是单机模式还是sharding模式插入性能都出现了陡降。

对于单机模式来说，达到2000万条性能骤减是由于服务器内存恰好全部占满，MongDB的内存映射方式使得当所有数据都在内存上的时候插入速度飞快，但是当部分数据需要移动到磁盘之后插入的性能下降严重。因为当集合中数据量过大时，MongDB会不断将磁盘的数据换入内存，造成IO压力非常大。

MongoDB 单集合索引测试

如上图所示，随着MongDB单集合的数据量增加，MongoDB的索引大小几乎也是直线型上升的，当数据量大概在2亿的时候，几乎占满了整个机器的内存。

显然易见的是，当MongoDB单表数据量过大的时候，索引更新更加困难，此时，MongoDB不仅要处理数据写入磁盘的工作还需要更新巨大的索引，这无疑也会对数据插入性能造成严重影响。

优化方法

明白了消息插入效率变低的原因之后，优化方法其实非常简单，只要控制MongoDB集群中的单表大小即可。

我们的项目中，一般一个主题是对应的一个工厂中的某类IOT数据，这些数据通常会被写入同一个名字的MongoDB 集合当中，而这个集合通常按照时间来拆分命令。

在优化之前，对于单表大概是以月为周期来进行表的拆分，这样的好处是方便根据时间查询历史数据的时候快速定位到相应的集合当中。但是随着项目的扩大，现在单表的数据量已经达到了数亿条的级别，无疑这样的数据量的情况下再进行数据的插入是很不明智的行为。

因此，我们对MongoDB的表拆分的时间间隔进行了调整，大概一周会拆分出一个新的表来进行数据的存储。对于历史数据，我们采用其它的办法来做查询优化，由于和Kafka的消息堆积无关这里我们不做讨论。优化之后，单表的数据量控制在几千万～1亿条之间，能够较好的提升插入效率。

在提升了插入性能的同时，Kafka的消费效率得到了一定程度上的提升，有效的缓解了消息积压的情况发生。

总结

这篇文章深入探讨了在智慧工厂IOT项目中，Kafka消息积压的优化方案。

首先，文章从业务场景、业务规模等方面介绍了该项目的背景，并详细说明了Kafka集群的构成，包括消息集群、Log集群、持久化集群和IOT数据集群。

在面对消息积压的挑战时，文章提出了一系列优化方案。通过IOT数据包的拆分和压缩，降低了消息体的大小，减轻了IO操作的负担，从而提高了Kafka的生产和消费效率。其次，通过优化消费者线程模型，采用单Kafka Consumer实例与多worker线程的模式，提高了整体的消费能力。文章还详细介绍了Kafka的消息拉取模型，并说明了为了保证消息的顺序性而采用的线程模型调整方案。

在处理MongoDB单表数据量过大的问题上，文章分析了插入性能受集合大小的影响，通过单表拆分和合理的索引管理等方式，有效提高了插入效率，缓解了Kafka集群的消息阻塞问题。

最后，文章总结了优化方案的核心思想，即通过合理的数据处理和管理策略，控制数据量大小，提高系统整体性能。这篇文章为处理大规模项目中Kafka消息积压问题提供了有力的实践经验和