（八）分佈式通訊----主機Host

 ==>>點擊查看本系列文章目錄

 

上節中有談到的是通訊主機（TransportHost），本節中主機（ServiceHost）負責管理服務的生命週期。

項目中將兩個主機拆分開，實現不一樣的功能：

通訊主機：用於啓動通訊監聽端口；

生命週期管理的主機：負責模塊功能的依賴注入，管理生命週期。

 

先看一下啓動服務端主機和客戶端主機後完成通訊的效果圖：

 

 文件結構以下：

 

ServiceHost 主機由ServiceHostBuilder來構建。

過程以下：

先看調用圖： 

1.Program中Main() 調用 ServiceHostBuilder 的方法：MapServices、RegisterServices、ConfigureServices、Configure

　　分別將委託填充到 List<Action<IContainer>>、List<Action<ContainerBuilder>>、List<Action<IServiceCollection>>、List<Action<IConfigurationBuilder>> 類型的容器中。

　　其中 IContainer、ContainerBuilder 是 Autofac中的容器，IServiceCollection、IConfigurationBuilder 是 Microsoft中的容器。

2. Program中Main() 調用 ServiceHostBuilder 的方法 UseStartup<Startup>()  ，Startup 必須實現 IStartup，完成Startup 的單例注入（微軟中的 Startup 能夠不實現 IStartup ，可是必須使用方法ConfigureServices、Configure）

3. Program中Main() 調用 ServiceHostBuilder 的方法 Build()

　　（1）回調容器 List<Action<IContainer>>、List<Action<ContainerBuilder>>、List<Action<IServiceCollection>>、List<Action<IConfigurationBuilder>> 中的委託。

　　　　  容器生成過程： ConfigureServices　　 ---》　　List<Action<IServiceCollection>>　　　　　　---》　　IServiceCollection

　　　　　　　　　　　　Configure　　　　　　---》　　List<Action<IConfigurationBuilder>>  　　　　---》　　IConfigurationBuilder

　　　　　　　　　　　　IServiceCollection + IConfigurationBuilder　　---》　　IServiceCollection 　　---》　　ServiceProvider

　　　　　　　　　　　　RegisterServices　　   ---》　　List<Action<ContainerBuilder>>　　　　　　 ---》　　ContainerBuilder

　　　　　　　　　　　　ContainerBuilder + IServiceCollection　　       ---》　　ContainerBuilder 

　　　　　　　　　　　　MapServices　　　　  ---》　　List<Action<IContainer>>

　　（2）將上面紅色字體的對象經過構造函數傳給new 的 ServiceHost 對象。

　　（3）調用ServiceHost .Initialize(),  該方法中執行以下過程

　　　　a. 用ServiceProvider 解析出 Startup 對象

　　　　b. 回調Startup的 IContainer ConfigureServices(ContainerBuilder builder) , 返回構建好的容器 IContainer

　　　　c. 回調Startup的 void Configure(IContainer app) , IContainer中注入其它功能

　　（4）將包含IContainer容器的ServiceHost 對象返回

4. ServiceHost.Run(), 回調主機中一直未執行的容器委託 List<Action<IContainer>> 

總結一下，整個過程就是將原來的四個委託的容器最後合併成一個 IContainer 容器。

解析容器中的服務，能夠用 ：

IContainer _container;
IDemoService service = _container.Resolve<IDemoService>(); 

 

服務端和客戶端啓動：

 

 

代碼：

咱們先看客戶端和服務端代碼：

服務端：

namespace Leo.ServiceLaunch.Server
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Server, Hello World!");

            var host = new ServiceHostBuilder()
                .RegisterServices(builder =>
                {
                    builder.RegisterType<MessagePackTransportMessageCodecFactory>().As<ITransportMessageCodecFactory>().SingleInstance();
                    builder.RegisterType(typeof(HttpServiceExecutor)).As(typeof(IServiceExecutor)).Named<IServiceExecutor>("tcp").SingleInstance();
                    builder.Register(provider =>
                    {
                        return new DotNettyServerMessageListener(provider.Resolve<ILogger<DotNettyServerMessageListener>>(),
                              provider.Resolve<ITransportMessageCodecFactory>());
                    }).SingleInstance();
                    builder.Register(provider =>
                    {
                        var serviceExecutor = provider.ResolveKeyed<IServiceExecutor>("tcp");
                        var messageListener = provider.Resolve<DotNettyServerMessageListener>();
                        return new DotNettyTransportHost(async endPoint =>
                        {
                            await messageListener.StartAsync(endPoint);
                            return messageListener;
                        }, serviceExecutor);
                    }).As<ITransportHost>();
                })
                .UseServer()  // 指定監聽的端口
                .UseStartup<Startup>()
                .Build();

            using (host.Run())
            {
                Console.WriteLine($"服務端啓動成功，{DateTime.Now}。");
            }

            Console.ReadLine();
        }

    }
}

namespace Leo.ServiceLaunch.Server
{
    class Startup : IStartup
    {
        public IContainer ConfigureServices(ContainerBuilder builder)
        {
            return builder.Build();
        }

        public void Configure(IContainer app)
        {

        }
    }
}

客戶端：

namespace Leo.ServiceLaunch.Client
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Client, Hello World!");

            var host = new ServiceHostBuilder()
                .RegisterServices(builder =>
                {
                    builder.RegisterType<MessagePackTransportMessageCodecFactory>().As<ITransportMessageCodecFactory>().SingleInstance();
                    builder.Register(provider =>
                    {
                        IServiceExecutor serviceExecutor = null;  
                        if (provider.IsRegistered(typeof(IServiceExecutor)))  // 沒有註冊客戶端接收消息執行器，所以一直爲空
                            serviceExecutor = provider.Resolve<IServiceExecutor>();
                        return new DotNettyTransportClientFactory(provider.Resolve<ITransportMessageCodecFactory>(),
                            provider.Resolve<ILogger<DotNettyTransportClientFactory>>(),
                            serviceExecutor);
                    }).As(typeof(ITransportClientFactory)).SingleInstance();
                })
                .UseStartup<Startup>()
                .Build();

            using (host.Run())
            {
                Startup.Test();
            }
            Console.ReadLine();
        }
    }
}

namespace Leo.ServiceLaunch.Client
{
    class Startup : IStartup
    {
        private static IContainer _container;
        public void Configure(IContainer app)
        {
        }

        public IContainer ConfigureServices(ContainerBuilder builder)
        {
            _container = builder.Build();
            return _container;
        }

        internal static void Test()
        {
            Task.Run(async () =>
            {
                do
                {
                    Console.WriteLine("正在循環 1萬次發送消息.....");

                    //1w次調用
                    var watch = Stopwatch.StartNew();
                    for (var i = 1; i < 10000; i++)
                    {
                        var invokeMessage = new TransportMessage
                        {
                            Id = i.ToString(),
                            ContentType = "string",
                            Content = "你好啊，這是客戶端發給服務端的消息"
                        };
                        try
                        {
                            var endPoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 981);
                            ITransportClientFactory transportClientFactory = _container.Resolve<ITransportClientFactory>();
                            var client = await transportClientFactory.CreateClientAsync(endPoint);
                            await client.SendAsync(invokeMessage);
                        }
                        catch (Exception exception)
                        {
                            Console.WriteLine(exception.ToString(), $"發起請求中發生了錯誤，服務Id：{invokeMessage.Id}。");
                            throw;
                        }
                    }
                    watch.Stop();
                    Console.WriteLine($"1萬次發送結束，執行時間：{watch.ElapsedMilliseconds}ms");
                    Console.WriteLine("Press any key to continue, q to exit the loop...");
                    var key = Console.ReadLine();
                    if (key.ToLower() == "q")
                        break;
                } while (true);
            }).Wait();
        }
    }
}

主機：

IServiceHost：

    public interface IServiceHost : IDisposable
    {
        IDisposable Run();

        IContainer Initialize();
    }

IServiceHostBuilder：

    public interface IServiceHostBuilder
    {
        IServiceHost Build();

        IServiceHostBuilder RegisterServices(Action<ContainerBuilder> builder);

        IServiceHostBuilder ConfigureServices(Action<IServiceCollection> configureServices);

        IServiceHostBuilder Configure(Action<IConfigurationBuilder> builder);

        IServiceHostBuilder MapServices(Action<IContainer> mapper);
    }

ServiceHost：

public class ServiceHost : IServiceHost
    {
        private readonly ContainerBuilder _builder;
        private IStartup _startup;
        private IContainer _applicationServices;
        private readonly IServiceProvider _hostingServiceProvider;
        private readonly List<Action<IContainer>> _mapServicesDelegates;

        public ServiceHost(ContainerBuilder builder,
            IServiceProvider hostingServiceProvider,
             List<Action<IContainer>> mapServicesDelegate)
        {
            _builder = builder;
            _hostingServiceProvider = hostingServiceProvider;
            _mapServicesDelegates = mapServicesDelegate;
        }

        public IContainer Initialize()
        {
            if (_applicationServices == null)
            {
                try
                {
                    if (_applicationServices == null)
                    {
                        if (_startup == null)
                        {
                            // 解析出 Startup 
                            _startup = _hostingServiceProvider.GetRequiredService<IStartup>();
                        }
                        //回調Startup中的 ConfigureServices，
                        _applicationServices = _startup.ConfigureServices(_builder);
                    }
                    if (_applicationServices == null)
                        _applicationServices = _builder.Build();
                    Action<IContainer> configure = _startup.Configure;
                    configure(_applicationServices);
                }
                catch (Exception ex)
                {
                    Console.Out.WriteLine("應用程序啓動異常: " + ex.ToString());
                    throw;
                }
            }
            return _applicationServices;
        }

        public IDisposable Run()
        {
            RunAsync().GetAwaiter().GetResult();
            return this;
        }

        public async Task RunAsync()
        {
            if (_applicationServices != null)
                MapperServices(_applicationServices);
        }

        private void MapperServices(IContainer mapper)
        {
            foreach (var mapServices in _mapServicesDelegates)
            {
                mapServices(mapper);
            }
        }

        public void Dispose()
        {
            (_hostingServiceProvider as IDisposable)?.Dispose();
        }
    }

ServiceHostBuilder：

public class ServiceHostBuilder : IServiceHostBuilder
    {
        private readonly List<Action<IServiceCollection>> _configureServicesDelegates;
        private readonly List<Action<ContainerBuilder>> _registerServicesDelegates;
        private readonly List<Action<IConfigurationBuilder>> _configureDelegates;
        private readonly List<Action<IContainer>> _mapServicesDelegates;

        public ServiceHostBuilder()
        {
            _configureServicesDelegates = new List<Action<IServiceCollection>>();
            _registerServicesDelegates = new List<Action<ContainerBuilder>>();
            _configureDelegates = new List<Action<IConfigurationBuilder>>();
            _mapServicesDelegates = new List<Action<IContainer>>();

        }

        public IServiceHost Build()
        {
            #region Microsoft原生的容器
            //執行 IServiceCollection 類型的委託
            var services = BuildCommonServices();
            //執行 IConfigurationBuilder 類型的委託
            var config = Configure();
            //日誌注入到 IServiceCollection
            services.AddLogging();
            //IConfigurationBuilder 注入到 IServiceCollection
            services.AddSingleton(typeof(IConfigurationBuilder), config);
            //用 IServiceCollection 生成 ServiceProvider 服務提供器
            var hostingServiceProvider = services.BuildServiceProvider();
            #endregion

            #region Autofac的容器
            //執行 ContainerBuilder 類型的委託
            var hostingServices = RegisterServices();
            #endregion

            //將 IServiceCollection 填充到 Autofac 的 ContainerBuilder 構建器中
            hostingServices.Populate(services);


            //把Autofac的ContainerBuild的容器構建器、Microsoft的ServiceProvider服務提供器、已有的IContainer容器的委託 都放入主機中
            var host = new ServiceHost(hostingServices, hostingServiceProvider, _mapServicesDelegates);
            //主機初始化之後返回的是IContainer容器
            var container = host.Initialize();
            return host;
        }

        public IServiceHostBuilder MapServices(Action<IContainer> mapper)
        {
            if (mapper == null)
            {
                throw new ArgumentNullException(nameof(mapper));
            }
            _mapServicesDelegates.Add(mapper);
            return this;
        }

        public IServiceHostBuilder RegisterServices(Action<ContainerBuilder> builder)
        {
            if (builder == null)
            {
                throw new ArgumentNullException(nameof(builder));
            }
            _registerServicesDelegates.Add(builder);
            return this;
        }

        public IServiceHostBuilder ConfigureServices(Action<IServiceCollection> configureServices)
        {
            if (configureServices == null)
            {
                throw new ArgumentNullException(nameof(configureServices));
            }
            _configureServicesDelegates.Add(configureServices);
            return this;
        }

        public IServiceHostBuilder Configure(Action<IConfigurationBuilder> builder)
        {
            if (builder == null)
            {
                throw new ArgumentNullException(nameof(builder));
            }
            _configureDelegates.Add(builder);
            return this;
        }

        private IServiceCollection BuildCommonServices()
        {
            var services = new ServiceCollection();
            foreach (var configureServices in _configureServicesDelegates)
            {
                configureServices(services);
            }
            return services;
        }

        private IConfigurationBuilder Configure()
        {
            //var config = new ConfigurationBuilder().SetBasePath(AppContext.BaseDirectory);
            var config = new ConfigurationBuilder();
            foreach (var configure in _configureDelegates)
            {
                configure(config);
            }
            return config;
        }

        private ContainerBuilder RegisterServices()
        {
            var hostingServices = new ContainerBuilder();
            foreach (var registerServices in _registerServicesDelegates)
            {
                registerServices(hostingServices);
            }
            return hostingServices;
        }
    }

IStartup：

    public interface IStartup
    {
        IContainer ConfigureServices(ContainerBuilder builder);

        void Configure(IContainer app);
    }

ServerExtensions：

    public static class ServerExtensions
    {
        public static IServiceHostBuilder UseServer(this IServiceHostBuilder hostBuilder)
        {
            return hostBuilder.MapServices(async mapper =>
            {
                int _port = 981;
                string _ip = "127.0.0.1";

                Console.WriteLine($"準備啓動服務主機，監聽地址：{_ip}:{_port}。");
                var transportHosts = mapper.Resolve<IList<ITransportHost>>();
                Task.Factory.StartNew(async () =>
                {
                    foreach (var transportHost in transportHosts)
                        await transportHost.StartAsync(_ip, _port);
                }).Wait();
            });
        }

        public static IServiceHostBuilder UseStartup<TStartup>(this IServiceHostBuilder hostBuilder) where TStartup : IStartup
        {
            return hostBuilder
                .ConfigureServices(services =>
                {
                    services.AddSingleton(typeof(IStartup), typeof(TStartup));
                });
        }
    }