基于Akka的高可靠分布式控制系统设计与实现

2022-02-25吕海东

微型电脑应用 2022年2期

关键词：线程参与者分布式

吕海东

(大连理工大学, 城市学院，辽宁, 大连 116600)

0 引言

当今世界，随着监测设备和控制设备日益大规模增加，尤其是物联网(IoT)的迅猛发展[1]，工业过程控制系统变得越来越复杂，已经由传统的集中控制模式，全面转变为分布式控制模式。

传统的控制系统通常可以控制几百或几千的传感器和执行设备，再多就感到力不从心，系统性能严重下降，实时性难以满足实际需求。如一个大城市的交通信号控制系统，智慧城市控制系统，需要控制的各种设备可能超过几十万，甚至上百万。

为了克服多线程编程复杂，难以实现真正的高并发性，高可靠性和实时性的问题，Lightbend推出了著名的Akka框架[2]。Akka为构建分布式、异步、高性能的实时系统提供了一种创新的编程模式模型。Akka使开发人员能够轻松地使用高级抽象来构建异步响应式系统，不必处理诸如线程、互斥和死锁之类难题。Akka将应用程序的行为和状态封装和建模为Actor(角色)，如过程控制系统中的每个设备对应一个Actor，这些Actor通过传递消息进行相互协作，完成数据的采集、传输和处理[3]。

Actor的核心工作模式是，应用程序只通过相互发送消息与之交互，而不像传统的面向对象编程模式直接调用对象的方法。

Akka的Actor对象占用资源非常少，按照Akka的官方文档说明，Akka的Actor运行在Java的JVM中，1GB Heap的内存可以运行250万个Actor对象，即可以模拟250万个控制设置(如传感器、执行器、转换器等)，通过增加计算机的内存，Akka可以控制的Actor数量急剧上升。

Akka中Actor的工作在消息驱动模式[4]，不像传统的编程模式通过调用对象的方法，或函数调用Actor功能，而是发送消息给Actor，Actor只有接收到消息后，才可以执行指定的功能。消息驱动完全工作在异步模式[5]，与传统的同步模式相比，没有线程的阻塞、锁等问题，极大提高了系统的处理性能。

1 系统总体架构设计

提出的过程控制系统基于Akka设计，当单个计算机无法满足控制的设备数量需求时，可以通过Akka-Cluster模块[6]，实现多个计算机的集群，将Akka的Actor System向外扩展并跨越许多计算机，从而实现完全分布式处理。

当监控设备增多，单个节点服务器无法创建足够多的参与者来映射每个设备，可采用Akka集群模块构建节点群。每个节点上构建参与者系统，在网络上监听各个Actor System跨节点的消息传输。集群上的参与者可以互相发送消息，处理来自分布式系统中设备的监测数据。

每个节点Akka参与者系统，负责控制和监视该节点的所有参与者。每个参与者映射到物联网设备，如温度传感器、压力传感器等设备，负责处理设备数据和执行设备(如流量控制阀等)的状态。

通过Akka MQTT连接器[7]，控制设备通过MQTT协议，通过发送消息将收集到的传感器数据发送到其对应的Actor参与者。参与者通过定义由邮箱接收消息事件触发的消息接收方法，并以异步响应式方式接收消息。Actor完全采用异步模型工作，不阻塞线程处理，可以处理大量的设备数据转换。

Akka中的Actor非常轻量级，每个只消耗几百个字节。根据Akka文档，拥有1GB内存的节点可以创建大约250万个Actor，而传统的控制系统采用多线程模式只能有1 024个线程，只能支持连接1 020个设备。

在Akka-Cluster集群模块的支持下，分布式控制系统的每个节点可以随着需要控制的设备的数量多少进行动态的添加或移除Actor System节点，从而使平台具有动态集群架构特性。基于Akka的控制系统总体架构如图1所示。

图1 基于Akka的控制系统总体架构图

2 系统核心控制功能的Actor层设计与实现

Actor模型使用发送和接收消息来实现设备间的通信和相互协作，有利于并发性和并行性，Actor是非常简单的抽象，有利于简化控制系统的编程。

Actor工作在异步、非阻塞和高性能的事件驱动编程模型中，所有Actor的工作必须通过给Actor发送消息实现。

为了在平台中呈现实际的设备，使用Scala开发了用于定义设备和设备类型的Actor类，下面的代码片段中展示了Actor对象的定义。

class DeviceType(name:String,mesureUnit:String){}

class Device(id:String,location:String,type:DeviceType){}

采集数据消息的设备是用Scala语言的case类实现的，当传感器设备通过MQTT协议向Actor系统发送数据时，内置的Akka MQTT连接器将流转换为Actor系统消息，该消息定义为以下代码片段。

case class SensorData(device:DeviceModel,data:Double);

传感器数据模型中包括采集数据的装置device及其测量数据值data，最后用device属性定义传感器Actor，并重写用接收到的消息触发的超类Actor方法receive。

在创建Actor类之后，在Actor系统中创建它的对象，Actor系统运行并监视所有Actor。下面的代码片段展示了如何使用Scala语言在Akka Actor系统中创建Actor对象，代码中201为设备的编号。

implicit val system = ActorSystem();

val sensor201=system.actorOf(Props[SensorActor],"NO201");

在Actor对象创建之后，它可以监视自己的邮箱，当邮箱中接收到消息时，事件触发器receive方法以响应式模式工作，不再阻塞线程。

在集群分布式系统中，参与者可以创建在远程节点的Actor系统中，Akka Remote模块实现参与者透明地向远程参与者发送消息。

3 分布式集群设计与实现

为了满足超大规模控制系统中设备和数据采集处理的需要，必须建立服务器集群。虽然Akka Actor非常轻量级，1GB HEAP的单台服务器可以容纳250万个Actor，可以支持相同数量的控制设备，但单个节点不支持容错，还需要集群架构。

Akka集群提供了一种基于点对点的容错分布式集群成员服务，没有单点故障或单点瓶颈。Akka集群允许构建分布式应用程序，其中一个应用程序或服务跨越多个节点(实际上是多个Actor系统)。

控制系统应用程序驻留在一个主节点中，主节点管理其他3个成员节点，这些节点可以动态加入集群。在每个节点上，应用程序类路径群集配置文件应用程序.conf创建为以下代码段。

akka {

cluster {

seed-nodes = [

"akka.tcp://IoTPlatform@210.30.108.224:2551",

"akka.tcp://IoTPlatform@210.30.108.225:2551",

"akka.tcp://IoTPlatform@210.30.108.226:2551"

]

}

akka.extensions=["akka.cluster.metrics.ClusterMetricsExtension"]

akka.cluster.metrics.native-library-extract-folder=${user.dir}/target/native

在Akka的集群系统中，先创建种子节点，再创建普通节点。当新普通节点启动时，它向所有种子节点发送一条消息，实现与种子节点交互。

4 控制系统采集和控制信息的发送

在控制系统设备采集层，采用JavaScript框架Johnny Five框架连接传感器等设备[8]，传感器数据通过Johnny Five获取后，发送到Kafka服务器[9]，并转换到Actor邮箱。

关键是以Apache-Kafka[10]为中心来接收传感器的监测数据，并向执行器发送动作指令。Apache-Kafka是一个发布/订阅消息传递系统，设计用于分布式集群环境中，Kafka中的数据被持久地、有序地存储，并且可以被确定地读取。Kafka常用于实时流数据体系结构，实现数据的实时处理。

对于设备层，当数据被获取时，发送到Kafka消息主题，Kafka节点是一个节点.js客户端，与Apache Kafka 集成。

采集的数据发送到Kafka后，本控制系统使用Alpakka-Kafka连接器连接Apache-Kafka，并将Kafka队列中的数据转换为Akka-Stream。Alpakka是一个开源项目，用于实现流数据的自动感知和响应式集成数据流管道，它建立在Akka Streams之上，为反应式和面向流的编程提供DSL，同时内置了对背压的支持，实现高可靠性的数据传输。

最后Alpakka消费者订阅Kafka主题并将消息传递到Akka流中。为了连接Kafka，ActorSystem内的ConsumerSettings配置为Akka-Kafka，同时消息的消费者在应用程序.conf文件配置。将Actor被编程为Kafka client，在接收事件时对消息做出反应，然后将Akka消息发送给Webs Socket Actor，将数据发送给Web客户端显示。