[ThingsBoard] 3. 源码解读Actor

一、前言

本文基于 ThingsBoard 4.0.2 编写，对应提交Version set to 4.0.2(01c5ba7d37006e1f8a3492afbb3c67d017ca8dd3)。
由于个人技术能力和写作经验有限，欢迎读者指出文中的错误与不足。
二、Actor模型

参考 Actor模型是解决高并发的终极解决方案 - 知乎
本人写的一般，可以看参考文章

1. Actor

Actor 模型则将一切视为 Actor。Actor 是并发执行的基本单元，与其他 Actor 之间通过消息传递进行通信，当接收到消息时，一个 Actor 可以并行地执行三件事：向其他 Actor 发送消息、创建新的 Actor，以及决定如何处理下一条消息。
2. 消息传递

Actor之间互相并行，因此消息传递都是异步的，Mailbox负责给Actor进行消息传递。
两个Actor之间消息传递时，Actor A发送到Actor B的Mailbox。等到Actor B处理消息时。
它就会从自己的Mailbox中获取消息，进行处理。
三、ThingsBoard中的Actor

ThingsBoard以下简称TB，代码进行一些精简，方法使用lambda方法引用表示。

1. TbActorMailBox

在TB的Actor模型中，Mailbox对应的实现类是TbActorMailbox，其继承关系为TbActorMailbox→TbActorCtx→TbActorRef。值得注意的是，TbActorCtx和TbActorRef在整个系统中都只有TbActorMailbox这一个实现类，因此在之后的代码阅读中可以将TbActorRef和TbActorCtx都认为是TbActorMailbox。这点在之后的代码分析很有用。
Actor通过其::tell和::tellWithHighPriority传递消息其对应的Actor。每个TbActor都只对应一个TbActorMailbox

1. private final TbActor actor;
3. public void tell(TbActorMsg actorMsg) {
4.     enqueue(actorMsg, NORMAL_PRIORITY);
5. }
7. public void tellWithHighPriority(TbActorMsg actorMsg) {
8.     enqueue(actorMsg, HIGH_PRIORITY);
9. }
11. private void tryProcessQueue(boolean newMsg) {
12.     dispatcher.getExecutor().execute(this::processMailbox);
13. }
16. private void enqueue(TbActorMsg msg, boolean highPriority) {
17.     if (highPriority) {
18.         highPriorityMsgs.add(msg);
19.     } else {
20.         normalPriorityMsgs.add(msg);
21.     }
22.     tryProcessQueue();
23. }

复制代码

其中::tryProcessQueue会异步处理消息队列，实际的处理逻辑在::processMailbox中。通过actor::process委托给实际的Actor执行。

1. private void processMailbox() {
2.     for (int i = 0; i < settings.getActorThroughput(); i++) {
3.         TbActorMsg msg = highPriorityMsgs.poll();
4.         if (msg == null) {
5.             msg = normalPriorityMsgs.poll();
6.         }
7.         if (msg != null) {
8.             actor.process(msg);
9.         }
10.     }
11. }

复制代码

2.Actor的初始化：

初始化的逻辑只选择部分重点分析，其他部分可以让ai能进行逐行分析。

Actor可以分为两大类

• 管理其他Actor，负责创建其管理的Actor，传递消息。
  
• 逻辑处理，接收消息，将消息委托给processor字段类处理。
  

2.1 管理创建Actor的Actor

AppActor

首先TB也是一个Spring应用，因此Actor的创建也是从被Spring管理的Bean开始第一个的创建。其受到Spring管理的就是DefaultActorService类在::initActorSystem中创建了AppActor。
看向TbActorSystem::createRootActor方法，它需要一个TbActorCreator的实例，而TbActorCreator则是负责创建每个TbActor的对象。会调用::createActor。

1. @PostConstruct
2. public void initActorSystem() {
3.     appActor = system.createRootActor(APP_DISPATCHER_NAME, new AppActor.ActorCreator(actorContext));
4. }

复制代码

::createActor中创建了一个TbActorMailbox。
在TbActorSystem内部会逐步调用到::createActor(String, TbActorCreator, TbActorId)，它加锁进行创建。

1. public TbActorRef createRootActor(String dispatcherId, TbActorCreator creator) {
2.     return createActor(dispatcherId, creator, null);
3. }
5. public TbActorRef createChildActor(String dispatcherId, TbActorCreator creator, TbActorId parent) {
6.     return createActor(dispatcherId, creator, parent);
7. }
9. private TbActorRef createActor(String dispatcherId, TbActorCreator creator, TbActorId parent) {
10.     Dispatcher dispatcher = dispatchers.get(dispatcherId)
11.     TbActorId actorId = creator.createActorId();
12.     TbActorMailbox actorMailbox = actors.get(actorId);
13.     Lock actorCreationLock = actorCreationLocks.computeIfAbsent(actorId, id -> new ReentrantLock());
14.     actorCreationLock.lock();
15.     try {
16.     if (actorMailbox == null) {
17.         log.debug("Creating actor with id [{}]!", actorId);
18.         TbActor actor = creator.createActor();
19.         TbActorRef parentRef = null;
20.         if (parent != null) {
21.             parentRef = getActor(parent);
22.         }
23.         TbActorMailbox mailbox = new TbActorMailbox(this, settings, actorId, parentRef, actor, dispatcher);
24.         actors.put(actorId, mailbox);
25.         mailbox.initActor(); // 注意此处
26.         actorMailbox = mailbox;
27.     } else {
28.         log.debug("Actor with id [{}] is already registered!", actorId);
29.     }
31.     } finally {
32.         actorCreationLock.unlock();
33.         actorCreationLocks.remove(actorId);
34.     }
35.     return actorMailbox;
36. }
38. private TbActorRef createActor(String dispatcherId, TbActorCreator creator, TbActorId parent) {
39.     TbActor actor = creator.createActor();
40.     TbActorRef parentRef = null;
41.     if (parent != null) {
42.         parentRef = getActor(parent);
43.     }
44.     TbActorMailbox mailbox = new TbActorMailbox(this, settings, actorId, parentRef, actor, dispatcher);
45.     actors.put(actorId, mailbox);
46.     mailbox.initActor();
47. }

复制代码

而在TbActorMailbox::initActor方法中，会提交::tryInit的异步初始化，其中actor.init(this)。就进行了Actor的初始化。

1. public void initActor() {
2.     dispatcher.getExecutor().execute(() -> tryInit(1));
3. }
5. private void tryInit(int attempt) {
6.     actor.init(this);
7. }

复制代码

TenantActor

AppActor::doProcess的第一个if，调用::initTenantActors,它初始化所有TenantActor。
AppActor自己管理着所有的TenantActors。其::getOrCreateTenantActor则完成了获取和创建的工作。
实现了一个懒加载的功能。

1. protected boolean doProcess(TbActorMsg msg) {
2.     if (!ruleChainsInitialized) {
3.         if (MsgType.APP_INIT_MSG.equals(msg.getMsgType())) {
4.             initTenantActors();
5.             ruleChainsInitialized = true;
6.         }
7.     }
8. }
10. private void initTenantActors() {
11.     PageDataIterable<Tenant> tenantIterator = new PageDataIterable<>(tenantService::findTenants, ENTITY_PACK_LIMIT);
12.     for (Tenant tenant : tenantIterator) {
13.         log.debug("[{}] Creating tenant actor", tenant.getId());
14.         getOrCreateTenantActor(tenant.getId())
15.     }
16. }

复制代码

追溯::getOrCreateTenantActor，他会使用类型为TbActorCtx的ctx字段的getOrCreateChildActor，根据之前的分析我们可以直接到TbActorMailbox进行分析。

1. private Optional<TbActorRef> getOrCreateTenantActor(TenantId tenantId) {
2.     return Optional.ofNullable(ctx.getOrCreateChildActor(new TbEntityActorId(tenantId),
3.             () -> DefaultActorService.TENANT_DISPATCHER_NAME,
4.             () -> new TenantActor.ActorCreator(systemContext, tenantId),
5.             () -> true));
6. }

复制代码

显示从类型为TbActorSystem的system字段获取actor。实现上DefaultTbActorSystem内有一个actors的ConcurrentMap的字段用于存储系统中所有的Actor。之后就和AppActor创建处一致的执行流程。

1. public TbActorRef getOrCreateChildActor(TbActorId actorId, Supplier<String> dispatcher, Supplier<TbActorCreator> creator, Supplier<Boolean> createCondition) {
2.     TbActorRef actorRef = system.getActor(actorId);
3.     if (actorRef == null && createCondition.get()) {
4.         return system.createChildActor(dispatcher.get(), creator.get(), selfId);
5.     } else {
6.         return actorRef;
7.     }
8. }

复制代码

RuleChainActor

同理RuleChainActor也是由TenantActor进行创建的。不过实际的逻辑在TenantActor的父类RuleChainManagerActor中。
一般都是在::init方法中调用::initRuleChains方法.initRuleChains方法就会从数据库中获取所有RuleChain并初始化。
2.2 负责逻辑处理的Actor

RuleChainActor

到了它，明显有了不一样，一下子类一下子变得简洁了。类本身只有两个方法
分析一下它的父类。RuleChainActor→RuleEngineComponentActor→ComponentActor。
其中的ComponentActor::init方法就调用了抽象方法ComponentActor::createProcessor。

1. @Override
2. public void init(TbActorCtx ctx) throws TbActorException {
3.     this.processor = createProcessor(ctx);
4.     initProcessor(ctx);
5. }

复制代码

再向上看ComponentActor的父类ContextAwareActor就发现，::process变成对::process做了个封装。只是添加了打印日志的功能。::doProcess其中，只是将对应的消息委托给了对应的processor。

1. public boolean process(TbActorMsg msg) {
2.     if (log.isDebugEnabled()) {
3.         log.debug("Processing msg: {}", msg);
4.     }
5.     if (!doProcess(msg)) {
6.         log.warn("Unprocessed message: {}!", msg);
7.     }
8.     return false;
9. }

复制代码

RuleNodeActor

这就是规则引擎最核心的部分，可以明显的推测到每个TbNode都对应着一个RuleNodeActor。
不过同理，它的实际逻辑也委托给了processor。但是明显它的创建更加复杂。
它的创建实际上是由RuleChainActor中的processor负责的，即RuleChainActorMessageProcessor类。

1. public void init(TbActorCtx ctx)  {
2.     this.processor = createProcessor(ctx);
3.     initProcessor(ctx);
4. }
6. protected void initProcessor(TbActorCtx ctx) {
7.     processor.start(ctx);
8. }

复制代码

其中的::start方法，而::start的方法则是在前文TbActor的父类ComponentActor::init中再调用::initProcessor中，
调用的RuleChainActorMessageProcessor::start方法。

1. public void start(TbActorCtx context) {
2.     List<RuleNode> ruleNodeList = service.getRuleChainNodes(tenantId, entityId);
3.     for (RuleNode ruleNode : ruleNodeList) {
4.         TbActorRef ruleNodeActor = createRuleNodeActor(context, ruleNode);
5.         nodeActors.put(ruleNode.getId(), new RuleNodeCtx(tenantId, self, ruleNodeActor, ruleNode));
6.     }
7.     initRoutes(ruleChain, ruleNodeList);
8. }

复制代码

其中从数据库中读取到所有nodeActors的信息并将其全部创建。
再调用::initRoutes初始化ruleChain中对应的路由。

1. private void initRoutes(RuleChain ruleChain, List<RuleNode> ruleNodeList) {
2.     for (RuleNode ruleNode : ruleNodeList) {
3.         List<EntityRelation> relations = service.getRuleNodeRelations(TenantId.SYS_TENANT_ID, ruleNode.getId());
4.         log.trace("[{}][{}][{}] Processing rule node relations [{}]", tenantId, entityId, ruleNode.getId(), relations.size());
5.         if (relations.isEmpty()) {
6.             nodeRoutes.put(ruleNode.getId(), Collections.emptyList());
7.         } else {
8.             for (EntityRelation relation : relations) {
9.                 nodeRoutes.computeIfAbsent(ruleNode.getId(), k -> new ArrayList<>())
10.                         .add(new RuleNodeRelation(ruleNode.getId(), relation.getTo(), relation.getType()));
11.             }
12.         }
13.     }
14. }

复制代码

能看出nodeRoutes接近一个邻接表的结构，构建了整个路由。至此将Actor创建的三种模式介绍清楚。

• 由外部创建，如AppActor。
  
• 由其父Actor创建，如TenantActor。
  
• 由对应的一个processor字段创建，如RuleChainActor。
  

四、结尾

最开始分析的时候，我还不了解Actor模型，纯粹看着代码进行分析，惊叹于设计的巧妙。但是在QQ水群突然有人提及自己在改为了无锁的Actor模型，我搜索一番，原来这是一个很成熟的通用的设计了啊。自己写小项目还是喜欢Executor和Future一股脑的用。不知道下一篇要多久才更新了。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！