NWPC消息平台：消息代理服务
本文属于介绍 NWPC 消息平台 系列文章。
简介
CMA 目前使用的高性能计算机系统 CMA-PI 由三种节点组成：计算节点、前后处理节点和管理登录节点。
CMA-PI 内部使用高速计算网络连接，与外界网络隔离，仅有前后处理节点和管理登录节点与 CMA 局域网相连。
详情请查看去年为单位培训准备的材料《NWPC高性能计算机环境介绍：高性能计算机概况》。
计算节点属于 CMA-PI 作业调度系统 Slurm 中的并行队列 (operation, normal, largemem, …)，前后处理节点属于串行队列 (serial_op, serial, …)。 数值预报业务系统的 ecFlow 服务程序运行在管理登录节点上 (login_b01 和 login_b06)，大部分任务会被提交到调度系统的并行队列和串行队列。 NWPC 消息平台使用的消息中间件运行在与 CMA 局域网相连的气象大数据平台中，无法与 CMA-PI 的高速计算网络进行通讯。 所以，需要有效的方式解决并行作业与消息中间件的通讯问题。
本文设计并实现一种消息代理服务，通过转发消息解决 CMA-PI 并行计算节点与 CMA 局域网无法通讯的问题。
设计
因为管理登录节点同时与 HPC 高速计算网络和 CMA 局域网连接，本文在 HPC 的管理登录节点上运行后台服务程序，提供消息代理转发服务。 该服务接收从 HPC 计算节点通过高速计算网络发送的事件消息，并通过 CMA 局域网转发给消息中间件。
架构设计
消息代理服务的数据流如下图所示。 代理服务运行在 CMA-PI 的登录节点上，并在高速计算网络上开放端口用于接收消息。 计算节点上运行的作业脚本调用消息发送客户端程序，通过端口向代理服务发送消息，代理服务接收消息后会通过 CMA 局域网将消息转发给消息队列。
消息代理服务数据流图，代理服务部署在 HPC 登录节点中
接口设计
消息代理服务的数据接口由两部分组成：事件消息以及消息中间件的连接信息。 本文为不同的消息中间件开发不同的数据接口，每种数据接口都包含相同格式的表示事件消息的字节流字段 (Message)，并针对不同中间件的需求定义不同的连接目标字段 (Target)。 目前已为 RabbitMQ 和 Kafka 开发数据接口，如下图所示，其中 Kafka 接口支持多个 Broker 地址。
消息代理服务数据接口，左图是 RabbitMQ 消息接口，右图是 Kafka 消息接口
关键技术
消息通讯
本文基于远程过程调用技术 (RPC) 构建消息代理服务的数据接口，使用 gRPC 实现网络通讯服务，使用 protobuf 实现消息的序列化。
Protobuf 全称 Protocol Buffer，是 Google 开发的序列化开源库，支持多种编程语言，具有良好的扩展能力 [1]。 gRPC 也是由 Google 开发的远程过程调用开源库，提供多种编程接口。
本文的代理服务基于 gRPC 的 GOLANG 接口开发，使用协程处理每个请求。 协程是一种可以被挂起和恢复的子程序，可以在单个线程中同时执行多个协程，不需要进程切换和锁机制，执行效率更高，资源占用更小，因此使用协程实现代理服务可以有效提高服务的吞吐量，保证消息转发步骤的高并发。
超时与重连机制
为了保证消息代理稳定运行，对转发消息到消息中间件的步骤进行时间限制，超过时限的发送过程会被强制终止。 同时，为进一步提高消息发送成功率，代理服务使用消息重新发送机制。 重新连接消息中间件的时间限制会被逐步放宽，目前使用的时间限制策略是发送次数乘以 2 秒，最多发送 2 次。
在超时和重连机制的配合下，消息代理服务既能最大程度保障消息转发，又不会因为消息发送延迟，积压大量客户端连接协程，导致代理服务占用过多资源。
高可用：服务监控
消息代理服务部署在 CMA-PI 的登录节点上。 为了保证服务持续稳定运行，本文基于 ecFLow 构建 HPC 登录节点后台服务监控系统，定时检查服务程序是否正常，在检测到异常时自动重启服务。
详情请参看《ecFlow笔记：基于ecFlow构建后台服务监控系统》。
基于 ecFlow 的后台服务监控系统运行流程示意图
应用
业务系统发送消息的吞吐量在 1000 条/秒左右，经测试，使用 Kafka 的消息代理服务可以满足需求。
目前已在两个非核心系统中启用 ecFlow 命令消息，包括
- globalchartos
- swfdp
结论
本文使用部署在 HPC 登录节点上的代理服务作为消息中转，有效解决 HPC 计算节点无法与外界通讯的问题。 消息代理服务基于 gRPC 开发，支持高并发，每秒可以转发 1000 条以上的事件消息，可以满足业务系统的需求。
实现
数据接口
接口的基本组成部分，包括
- RabbitMQ 连接信息
RabbitMQTarget - Kafka 连接信息
KafkaTarget - 消息内容字节流
Message
message RabbitMQTarget {
string server = 1;
string exchange = 2;
string route_key = 3;
}
message KafkaTarget {
repeated string brokers = 1;
string topic = 2;
}
message Message {
bytes data = 1;
}
两个消息数据接口 RabbitMQMessage 和 KafkaMessage。
返回接口 Response。
message RabbitMQMessage {
RabbitMQTarget target = 1;
Message message = 2;
}
message KafkaMessage {
KafkaTarget target = 1;
Message message = 2;
}
message Response {
int32 error_no = 1;
string error_message = 2;
}
远程过程调用接口 SendRabbitMQMessage 和 SendKafkaMessage
service MessageBroker{
rpc SendRabbitMQMessage(RabbitMQMessage) returns (Response) {}
rpc SendKafkaMessage(KafkaMessage) returns (Response) {}
}
参考
参考文献
[1] 黄斌,谢艳新,唐友, 等.数据存储信息序列化完整性及效率评估仿真[J].计算机仿真,2020,37(4):159-163.
项目地址
