以下的分析环境基于默认配置

1. Actor 模型

首先搬用官网关于 Actor 模型的介绍：
ThingsBoard源码分析之启动分析 3 - 图1

  • The brief description of each actor’s functionality is listed below:
  • App Actor - responsible for management of tenant actors（负责管理租户 Actor）. An instance of this actor is always present in memory.
  • Tenant Actor - responsible for management of tenant device & rule chain actors（负责管理租户设备，规则链 Actor）. An instance of this actor is always present in memory.
  • Device Actor - maintain state of the device: active sessions, subscriptions, pending RPC commands（处理设备状态,激活 session,订阅，RPC 请求）, etc. Caches current device attributes in memory for performance reasons. An actor is created when the first message from the device is processed. The actor is stopped when there is no messages from devices for a certain time.
  • Rule Chain Actor - process incoming messages and dispatches them to rule node actors（处理消息，并将消息分发给 node Actor）. An instance of this actor is always present in memory.

  • Rule Node Actor - process incoming messages, and report results back to rule chain actor（处理消息，并将处理的消息返回给规则链 Actor）. An instance of this actor is always present in memory.

    2. Actor 启动分析

    ThingsBoard源码分析之启动分析 3 - 图2
    图片看不清楚的，请点击这里：

  • 标准：https://cdn.iotschool.com/photo/2020/633f6162-71be-409e-925e-762d51e29b19.png?x-oss-process=image/resize,w_1920

  • 高清：https://pic.downk.cc/item/5f50c6b1160a154a67353d2e.png

DefaultActorService中@PostConstruct标记的initActorSystem()开启了 actor 的启动流程：

  1. public void initActorSystem() {
  2.     log.info("Initializing actor system.");
  3.     actorContext.setActorService(this);
  4.     TbActorSystemSettings settings = new TbActorSystemSettings(actorThroughput, schedulerPoolSize, maxActorInitAttempts);
  5.     system = new DefaultTbActorSystem(settings);
  6.     //创建不同actor执行的线程池
  7.     system.createDispatcher(APP_DISPATCHER_NAME, initDispatcherExecutor(APP_DISPATCHER_NAME, appDispatcherSize));
  8.     system.createDispatcher(TENANT_DISPATCHER_NAME, initDispatcherExecutor(TENANT_DISPATCHER_NAME, tenantDispatcherSize));
  9.     system.createDispatcher(DEVICE_DISPATCHER_NAME, initDispatcherExecutor(DEVICE_DISPATCHER_NAME, deviceDispatcherSize));
  10.     system.createDispatcher(RULE_DISPATCHER_NAME, initDispatcherExecutor(RULE_DISPATCHER_NAME, ruleDispatcherSize));
  11.     actorContext.setActorSystem(system);
  12.     //创建AppActor详细分析见①createActor分析
  13.     appActor = system.createRootActor(APP_DISPATCHER_NAME, new AppActor.ActorCreator(actorContext));
  14.     actorContext.setAppActor(appActor);
  15.     //创建statsActor,暂时不关注
  16.     TbActorRef statsActor = system.createRootActor(TENANT_DISPATCHER_NAME, new StatsActor.ActorCreator(actorContext, "StatsActor"));
  17.     actorContext.setStatsActor(statsActor);
  18.     log.info("Actor system initialized.");
  19. }

①DefaultTbActorSystem类中方法createActor分析

  1. private TbActorRef createActor(String dispatcherId, TbActorCreator creator, TbActorId parent) {
  2.     //获取initActorSystem时对应的线程池
  3.     Dispatcher dispatcher = dispatchers.get(dispatcherId);
  4.     if (dispatcher == null) {
  5.         log.warn("Dispatcher with id [{}] is not registered!", dispatcherId);
  6.         throw new RuntimeException("Dispatcher with id [" + dispatcherId + "] is not registered!");
  7.     }
  8.     //creator有不同的实现如AppActor，TenantActor,RuleChainActor, RuleNodeActor等
  9.     // AppActor, 则使用固定的NULL_UUID,UUID.fromString("13814000-1dd2-11b2-8080-808080808080")作为Id
  10.     // TenantActor 使用creator提供的tenantId作为Id
  11.     // RuleChainActor 使用creator提供的ruleChain作为Id
  12.     // RuleNodeActor  使用creator提供的ruleNodeId作为Id
  13.     TbActorId actorId = creator.createActorId();
  14.     //缓存使用，将已经创建的放入缓存
  15.     TbActorMailbox actorMailbox = actors.get(actorId);
  16.     if (actorMailbox != null) {
  17.         log.debug("Actor with id [{}] is already registered!", actorId);
  18.     } else {
  19.         Lock actorCreationLock = actorCreationLocks.computeIfAbsent(actorId, id -> new ReentrantLock());
  20.         actorCreationLock.lock();
  21.         try {
  22.             actorMailbox = actors.get(actorId);
  23.             if (actorMailbox == null) {
  24.                 log.debug("Creating actor with id [{}]!", actorId);
  25.                 //正式创建Actor,也就是调用构造函数
  26.                 TbActor actor = creator.createActor();
  27.                 //只允许AppActor的parent是空
  28.                 TbActorRef parentRef = null;
  29.                 if (parent != null) {
  30.                     parentRef = getActor(parent);
  31.                     if (parentRef == null) {
  32.                         throw new TbActorNotRegisteredException(parent, "Parent Actor with id [" + parent + "] is not registered!");
  33.                     }
  34.                 }
  35.                 //使用actor创建TbActorMailbox对象
  36.                 TbActorMailbox mailbox = new TbActorMailbox(this, settings, actorId, parentRef, actor, dispatcher);
  37.                 //放入缓存
  38.                 actors.put(actorId, mailbox);
  39.                 //用各自的线程池去进行初始化操作，具体的初始化操作由每种actor的init方法实现详细分析如下②③④
  40.                 mailbox.initActor();
  41.                 actorMailbox = mailbox;
  42.                 if (parent != null) {
  43.                     //缓存，父子关系缓存
  44.                     parentChildMap.computeIfAbsent(parent, id -> ConcurrentHashMap.newKeySet()).add(actorId);
  45.                 }
  46.             } else {
  47.                 log.debug("Actor with id [{}] is already registered!", actorId);
  48.             }
  49.         } finally {
  50.             actorCreationLock.unlock();
  51.             actorCreationLocks.remove(actorId);
  52.         }
  53.     }
  54.     return actorMailbox;
  55. }
  • ② TenantActor的init方法：

如果此应用不是以某一个 tennant 启动的话或者isolatedTbRuleEngine是 false，调用父类RuleChainManagerActor的initRuleChains()方法查询该租户下的rule_chain表。开始创建RuleChainActor, 创建RuleChainActor的 parent 是当前TenantActor,创建的时候，又会调用到DefaultTbActorSystem的createActor方法；

  • ③RuleChainActor实现了ComponentActor， ComponentActor的init方法调用了createProcessor(TbActorCtx ctx)（由子类实现，创建了RuleChainActorMessageProcessor实例）和initProcessor(TbActorCtx ctx)（调用processor.start也即RuleChainActorMessageProcessor的start方法，该方法查询查询relation表，条件为 RULE_CHAIN 和 ruleChainId，再根据 to_id 去rule_node表获取RuleNode；1. 根据 ruleNodeList 开始初创建RuleNodeActor2.initRoutes()查询relation,rule_node，rule_chain表，构造 nodeRoutes 的数据，并设置 firstRuleNode）;
  • ④RuleNodeActor实现了ComponentActor， ComponentActor的init方法调用了createProcessor(TbActorCtx ctx)（由子类实现，创建了RuleNodeActorMessageProcessor实例）和initProcessor(TbActorCtx ctx),（调用processor.start也即RuleNodeActorMessageProcessor的start方法，使用反射创建 ruleNode 的实例，并按照数据库 configuration 的配置，调用init方法）;

DefaultActorService中@EventListener(ApplicationReadyEvent.class)标记的onApplicationEvent开始进一步初始化流程：
调用appActor.tellWithHighPriority(new AppInitMsg());将AppInitMsg推入AppActor的消息队列里，然后尝试处理这一消息，最终会交给AppActor的doProcess方法进行处理，由于还未ruleChainsInitialized从未进行复制，此时 AppActor 开始初始化 TennatActor,查询tenant表对所有 tenant 创建TenantActor;

总结

  1. initActorSystem创建AppActor和StatsActor两个 Actor,并未创建其他的 Actor;
  2. 在收到ApplicationReadyEvent的时候，由于之前没有初始化，所以会初始化该应用的所有TenantActor, 根据②③④点分析，循环切递归的创建了所有的RuleChainActor, RuleNodeActor;
  3. 关于TbActorMailbox的理解：mailbox 理解为一个信箱，里面有一些信件 (即入队列的消息)，这些信件有一些是高优先级，有些是普通优先级的，每次取信件的时候，都先看有没高优先级的，先处理高优先级的信件，再处理普通优先级的信件。那么信件处理人是谁呢？信件的处理人就是每一个 Actor.实际上的信件处理方法都是每一个 Actor 的doProcess(TbActorMsg msg)方法；
  4. 我们讨论的每一个RuleNodeActor就是在前端 RULE CHAINS 界面看到的每一个节点。