前言在上一篇文章中，我们提到车联网TSP平台有很多不同的业务主题，介绍了如何根据不同的业务主题设计MQTT业务场景.车辆会不断产生海量信息，通过车联网上报的每一条数据都非常宝贵，背后蕴藏着巨大的商业价值。因此，我们搭建的车载TSP平台通常需要具备千万级的topic和百万级的消息吞吐能力。传统的互联网系统难以支撑百万级订单的消息吞吐量。在本文中，我们将主要介绍如何设计下一代车联网平台的架构，以满足百万级消息吞吐量的需求。车联网场景消息吞吐量设计相关因素车联网消息分为上行链路和下行链路。上行消息一般是传感器和车辆的报警，将设备信息发送到云消息平台。下行消息一般包括遥控指令集消息和消息推送，云平台向车辆发送相应的指令。在车联网消息吞吐量的设计中，我们需要关注以下几个因素：(1)消息频率汽车在行驶过程中，GPS、车载传感器等都在不断地采集消息。为了接收到实时的反馈信息，他们上报接收消息也非常频繁。上报频率一般在100ms到30s之间，所以当车辆数量达到百万级时，平台需要支持百万级/秒的消息吞吐量。（2）消息包大小消息通过各种传感器采集自身的环境和状态信息（车联网场景常见新能源国标数据和企业标准数据）。整个消息包的大小一般在500B到几十KB之间。当同时上报大量报文包时，车联网平台需要有更强的接收和发送大报文包的能力。(3)消息延迟车辆在行驶过程中，消息数据只能通过无线网络传输。在大多数车联网场景下，车辆对时延的要求是ms级别的。平台还需要在满足百万级吞吐量条件的同时，保持低时延的消息传输。(4)主题的数量和层级在考虑百万级消息吞吐场景时，还需要规范消息主题的数量和主题树的层级设计。(5)Payload编解码当消息包比较大时，需要重点关注消息体的封装。简单的JSON封装对于消息解析来说效率不够。可以考虑使用Avro、Protobuf等编码格式进行Payload格式封装。对于百万级消息吞吐场景，传统的基于MQTT客户端共享订阅消息或者通过规则引擎实时写入关系数据库的架构显然不能令人满意。目前主流的架构选择有两种：一种是消息接入产品/服务+消息队列（Kafka、Pulsar、RabbitMQ、RocketMQ等），一种是消息接入产品/服务+时序数据库（InfluxDB、TDengine、Lindorm等）来实现。下面我们将基于上述相关因素和客户案例的最佳实践，采用云原生分布式物联网消息服务器EMQX作为消息接入层，分别介绍两种架构的实现方式。EMQX+Kafka构建百万级吞吐量车联网平台架构设计Kafka作为主流消息队列之一，具备持久化数据存储能力，可以进行持久化操作。同时通过数据持久化到硬盘和复制来防止数据丢失。后台TSP平台或大数据平台可以批量订阅想要的消息。因为Kafka具有订阅和发布的能力，可以接收来自南方的上报消息，并缓存起来；也可以将要发送的命令通过接口通过北接发送给前端进行命令下发。下面以Kafka为例搭建一个EMQX+Kafka百万级吞吐量的车联网平台：前端车机的连接和消息可以通过公有云提供的负载均衡产品做域名转发提供商。如果采用TLS/DTLS安全认证，可以在云端搭建4台HAProxy/Nginx服务器，进行证书分流和负载均衡。10个EMQX组成一个大集群，百万消息吞吐量平分到每个节点十万消息吞吐量，同时满足高可用场景的需求。如果有离线/消息缓存需求，可以使用Redis作为存储数据库。Kafka作为整体消息队列，EMQX通过规则引擎将全量消息转发到后端Kafka集群。后端TSP平台/OTA等应用通过订阅Kafka主题接收相应的消息，业务平台的控制指令和推送消息可以通过Kafka/API发送给EMQX。整体架构图在本方案架构中，EMQX作为消息中间件具有以下优势，可以满足该场景的需求：支持千万级车连接和百万级消息吞吐量。分布式集群架构，稳定可靠，支持动态水平扩展。强大的规则引擎和数据桥接持久化能力，支持百万级消息吞吐处理。丰富的API和认证体系，可以流畅对接。百万级吞吐量场景验证为了验证上述架构的吞吐能力，如果条件允许，我们可以通过如下配置搭建百万级消息吞吐量测试场景。压测工具可以选择BenchmarkTools、JMeter或者XMeter测试平台。一共模拟了100万台设备，每台设备都有自己的主题，每台设备每秒发送一条消息，压力测试持续12小时。压测架构图如下：性能测试结果呈现：(1)EMQX集群Dashboard统计EMQX规则引擎可以看到每个节点的处理速度为100,000/s，10个节点的总速度为100万/秒。.(2)EMQX规则引擎统计可以看到在Kafka中有每秒100万条的写入速度，并且一直在持续存储。Kafka管理接口统计EMQX+InfluxDB构建百万级吞吐量车联网平台架构设计采用EMQX+时序数据库架构，同样可以构建百万级消息吞吐量平台。在本文中，我们以InfluxDB时序数据库为例。InfluxDB是一款高性能时序数据库，广泛应用于存储系统监控数据、物联网行业实时数据等场景。从时间维度记录消息，具有强大的写入和存储性能，适用于大数据和数据分析。分析后的数据可以提供给后台应用系统进行数据支持。在该架构中，使用EMQX规则引擎进行消息转发，使用InfluxDB进行消息存储，并与后端大数据和分析平台对接，可以更方便的为时序分析服务。前端设备的报文通过云厂商的负载均衡产品进行域名转发和负载均衡。本次以1个EMQX作为测试，后期如果有需要可以采用多节点的方式组成相应的集群方案（100万的测试可以部署10个EMQX集群）。如果有离线/消息缓存需求，可以使用Redis作为存储数据库。EMQX通过规则引擎将消息全量转发到后端InfluxDB进行持久化数据存储。后端大数据平台通过InfluxDB接收相应的消息，对其进行大数据分析，分析后通过API将需要的信息传输给EMQX。总体架构图场景验证如测试架构图所示，XMeter压力机模拟10万个MQTT客户端向EMQX发起连接，新建连接速率为10000/秒，客户端心跳间隔（KeepAlive）为300秒。所有连接成功后，每个客户端发送一条QoS为1，Payload为每秒200B的消息。所有消息都经过HTTPInfluxDB规则引擎桥过滤，并以持久的方式发送到InfluxDB数据库。测试结果展示如下：EMQXDashboard统计：EMQX规则引擎统计：InfluxDB数据库接收数据：EMQXDashboard消息计数统计单台EMQX服务器已经实现了单台服务器10万TPS的消息吞吐量持久化到InfluxDB的能力。参考EMQX+Kafka架构的测试场景，将EMQX的集群节点扩展到10个可以支持100万TPS的消息吞吐量。结语通过本文，我们介绍了车联网场景下消息吞吐量设计需要考虑的因素，并提供了两种比较主流的百万级吞吐量平台架构设计方案。面对车联网场景的数据量越来越大，希望本文能为相关团队和开发者在车联网平台的设计和开发过程中提供参考。
