Akka.net路径里的user

Posted 2020-06-18 起名什么的...

  因为经常买双色球,嫌每次对彩票号麻烦,于是休息的时候做了个双色球兑奖的小程序,做完了发现业务还挺复杂的,于是改DDD重做设计,拆分服务,各种折腾...,不过这和本随笔没多大关系,等差不多了再总结一下,有空就写出来。

  上面说了拆服务，拆成了录入，开奖，兑奖三个服务，三个服务开始是用消息队列的，不过后来发现其实就是服务的领域对象之间的交互，就联想到了Actor，面向对象于是一切都是对象，actor也是一切都是actor，看着就门当户对，于是就干掉消息队列。。。细节以后有机会再说。

  上面其实也和这随笔没啥关系，有关系的是，因为时间关系，没去了解akka的文档，于是好奇一件事，服务器连接地址中的"user"是哪来的：

//var section = (AkkaConfigurationSection)ConfigurationManager.GetSection("akka");

//using (var system = ActorSystem.Create("TestClient", section.AkkaConfig))

            using (var system = ActorSystem.Create("TestClient"))
            {
                var actor = system.ActorSelection("akka.tcp://[email protected]:8081/user/tester");

                while (true)
                {
                    var input = Console.ReadLine();
                    if (input.Equals("1"))
                    {
                        actor.Tell(new DTO("11111"));
                    }
                    else if (input.Equals("2"))
                    {
                        actor.Tell(new DTO("22222"));
                    }
                    else
                    {
                        actor.Tell(new DTO("H W"));
                    }
                }
            }

    上面是客户端，下面是服务端：

 1             var config = ConfigurationFactory.ParseString(@"
 2             akka {  
 3                 actor {
 4                     provider = ""Akka.Remote.RemoteActorRefProvider, Akka.Remote""
 5                 }
 6                 remote {
 7                     helios.tcp {
 8                         transport-class = ""Akka.Remote.Transport.Helios.HeliosTcpTransport, Akka.Remote""
 9                         applied-adapters = []
10                         transport-protocol = tcp
11                         port = 8081
12                         hostname = localhost
13                     }
14                 }
15             }
16             ");
17 
18             using (var system = ActorSystem.Create("TestServer", config))
19             {
20                 system.ActorOf<TestDTOActor>("tester");
21 
22                 Console.ReadLine();
23             }

  文档想必很多，最近有别的要看，于是想想看看代码得了，反正一时还不会用到生产环境，到时候再说吧，先解决好奇的问题。

  于是首先是ActorSystem的实现类ActorSystemImpl，system.ActorOf<TestDTOActor>("tester")对服务端路径影响的应该是ActorOf：

        public override IActorRef SystemActorOf<TActor>(string name = null)
        {
            return _provider.SystemGuardian.Cell.ActorOf<TActor>(name);
        }

  于是找找_provider:

        public ActorSystemImpl(string name, Config config)
        {
            if(!Regex.Match(name, "^[a-zA-Z0-9][a-zA-Z0-9-]*$").Success)
                throw new ArgumentException(
                    "invalid ActorSystem name [" + name +
                    "], must contain only word characters (i.e. [a-zA-Z0-9] plus non-leading ‘-‘)");
            if(config == null)
                throw new ArgumentNullException("config");

            _name = name;            
            ConfigureSettings(config);
            ConfigureEventStream();
            ConfigureProvider();
            ConfigureTerminationCallbacks();
            ConfigureScheduler();
            ConfigureSerialization();
            ConfigureMailboxes();
            ConfigureDispatchers();
            ConfigureActorProducerPipeline();
        }

        private void ConfigureSettings(Config config)
        {
            _settings = new Settings(this, config);
        }

        private void ConfigureProvider()
        {
            Type providerType = Type.GetType(_settings.ProviderClass);
            global::System.Diagnostics.Debug.Assert(providerType != null, "providerType != null");
            var provider = (IActorRefProvider)Activator.CreateInstance(providerType, _name, _settings, _eventStream);
            _provider = provider;
        }

        public Settings(ActorSystem system, Config config)
        {
            _userConfig = config;
            _fallbackConfig = ConfigurationFactory.Default();            
            RebuildConfig();

            System = system;
            
            ConfigVersion = Config.GetString("akka.version");
            ProviderClass = Config.GetString("akka.actor.provider");
            var providerType = Type.GetType(ProviderClass);
            if (providerType == null)
                throw new ConfigurationException(string.Format("‘akka.actor.provider‘ is not a valid type name : ‘{0}‘", ProviderClass));
            if (!typeof(IActorRefProvider).IsAssignableFrom(providerType))
                throw new ConfigurationException(string.Format("‘akka.actor.provider‘ is not a valid actor ref provider: ‘{0}‘", ProviderClass));
            
            SupervisorStrategyClass = Config.GetString("akka.actor.guardian-supervisor-strategy");

            AskTimeout = Config.GetTimeSpan("akka.actor.ask-timeout", allowInfinite: true);
            CreationTimeout = Config.GetTimeSpan("akka.actor.creation-timeout");
            UnstartedPushTimeout = Config.GetTimeSpan("akka.actor.unstarted-push-timeout");

            SerializeAllMessages = Config.GetBoolean("akka.actor.serialize-messages");
            SerializeAllCreators = Config.GetBoolean("akka.actor.serialize-creators");

            LogLevel = Config.GetString("akka.loglevel");
            StdoutLogLevel = Config.GetString("akka.stdout-loglevel");
            Loggers = Config.GetStringList("akka.loggers");

            LoggerStartTimeout = Config.GetTimeSpan("akka.logger-startup-timeout");

            //handled
            LogConfigOnStart = Config.GetBoolean("akka.log-config-on-start");
            LogDeadLetters = 0;
            switch (Config.GetString("akka.log-dead-letters"))
            {
                case "on":
                case "true":
                    LogDeadLetters = int.MaxValue;
                    break;
                case "off":
                case "false":
                    LogDeadLetters = 0;
                    break;
                default:
                    LogDeadLetters = Config.GetInt("akka.log-dead-letters");
                    break;
            }
            LogDeadLettersDuringShutdown = Config.GetBoolean("akka.log-dead-letters-during-shutdown");
            AddLoggingReceive = Config.GetBoolean("akka.actor.debug.receive");
            DebugAutoReceive = Config.GetBoolean("akka.actor.debug.autoreceive");
            DebugLifecycle = Config.GetBoolean("akka.actor.debug.lifecycle");
            FsmDebugEvent = Config.GetBoolean("akka.actor.debug.fsm");
            DebugEventStream = Config.GetBoolean("akka.actor.debug.event-stream");
            DebugUnhandledMessage = Config.GetBoolean("akka.actor.debug.unhandled");
            DebugRouterMisconfiguration = Config.GetBoolean("akka.actor.debug.router-misconfiguration");
            Home = Config.GetString("akka.home") ?? "";
            DefaultVirtualNodesFactor = Config.GetInt("akka.actor.deployment.default.virtual-nodes-factor");

            SchedulerClass = Config.GetString("akka.scheduler.implementation");
            //TODO: dunno.. we dont have FiniteStateMachines, dont know what the rest is
            /*              
                final val SchedulerClass: String = getString("akka.scheduler.implementation")
                final val Daemonicity: Boolean = getBoolean("akka.daemonic")                
                final val DefaultVirtualNodesFactor: Int = getInt("akka.actor.deployment.default.virtual-nodes-factor")
             */
        }

  很明显，看配置文件的akka.actor.provider:

<configuration>
  <configSections>
    <section name="akka" type="Akka.Configuration.Hocon.AkkaConfigurationSection, Akka" />
  </configSections>
    <startup> 
        <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.5" />
    </startup>
  <akka>
    <hocon>
      <![CDATA[
          akka {  
                actor {
                    provider = "Akka.Remote.RemoteActorRefProvider, Akka.Remote"
                }
                remote {
                    helios.tcp {
                        transport-class = "Akka.Remote.Transport.Helios.HeliosTcpTransport, Akka.Remote"
                        applied-adapters = []
                        transport-protocol = tcp
                        port = 0
                        hostname = localhost
                    }
                }
            }
      ]]>
    </hocon>
  </akka>
</configuration>

  于是RemoteActorRefProvider：

        public LocalActorRef SystemGuardian { get { return _local.SystemGuardian; } }

        public RemoteActorRefProvider(string systemName, Settings settings, EventStream eventStream)
        {
            settings.InjectTopLevelFallback(RemoteConfigFactory.Default());

            var remoteDeployer = new RemoteDeployer(settings);
            Func<ActorPath, IInternalActorRef> deadLettersFactory = path => new RemoteDeadLetterActorRef(this, path, eventStream);
            _local = new LocalActorRefProvider(systemName, settings, eventStream, remoteDeployer, deadLettersFactory);
            RemoteSettings = new RemoteSettings(settings.Config);
            Deployer = remoteDeployer;
            _log = _local.Log;
        }

  好吧，看LocalActorRefProvider:

        public void Init(ActorSystemImpl system)
        {
            _system = system;
            //The following are the lazy val statements in Akka
            var defaultDispatcher = system.Dispatchers.DefaultGlobalDispatcher;
            _defaultMailbox = () => new ConcurrentQueueMailbox(); //TODO:system.Mailboxes.FromConfig(Mailboxes.DefaultMailboxId)
            _rootGuardian = CreateRootGuardian(system);
            _tempContainer = new VirtualPathContainer(system.Provider, _tempNode, _rootGuardian, _log);
            _rootGuardian.SetTempContainer(_tempContainer);
            _userGuardian = CreateUserGuardian(_rootGuardian, "user");
            _systemGuardian = CreateSystemGuardian(_rootGuardian, "system", _userGuardian);
            //End of lazy val

            _rootGuardian.Start();
            // chain death watchers so that killing guardian stops the application
            _systemGuardian.Tell(new Watch(_userGuardian, _systemGuardian));    //Should be SendSystemMessage
            _rootGuardian.Tell(new Watch(_systemGuardian, _rootGuardian));      //Should be SendSystemMessage
            _eventStream.StartDefaultLoggers(_system);
        }

        private LocalActorRef CreateUserGuardian(LocalActorRef rootGuardian, string name)   //Corresponds to Akka‘s: override lazy val guardian: LocalActorRef
        {
            return CreateRootGuardianChild(rootGuardian, name, () =>
            {
                var props = Props.Create<GuardianActor>(UserGuardianSupervisorStrategy);

                var userGuardian = new LocalActorRef(_system, props, DefaultDispatcher, _defaultMailbox, rootGuardian, RootPath/name);
                return userGuardian;
            });         
        }

  看样子，这个RootPath还重载了操作符：

        public static ActorPath operator /(ActorPath path, string name)
        {
            return new ChildActorPath(path, name, 0);
        }

        public static ActorPath operator /(ActorPath path, IEnumerable<string> name)
        {
            var a = path;
            foreach (string element in name)
            {
                a = a / element;
            }
            return a;
        }

   现在_userGuardian路径是看到了，生拼上去的，那么哪里用的呢

LocalActorRefProvider:

     public LocalActorRef Guardian { get { return _userGuardian; } }

RemoteActorRefProvider:

     public LocalActorRef Guardian { get { return _local.Guardian; } }

  再回来看system.ActorOf<TestDTOActor>("tester")这句:

        public override IActorRef ActorOf(Props props, string name = null)
        {
            return _provider.Guardian.Cell.ActorOf(props, name: name);
        }

  这就是我自己配的provider造成的，其实还是配置问题，当然，这么看我们其实可以实现自己的provider，想怎么搞就怎么搞，有时间其实还是应该看下文档，不过想来文档应该会说怎么用什么样，当应该不会说怎么实现的，要知道为什么，怎么改，估计还是要自己看代码，此致。