surging 集成SuperSocket预发布版本2.0

一、概述

周末在家试着扩展SuperSocket，因为之前都是只支持.net framework, 后面出现支持.NET CORE 的SuperSocket 2.0 ，然后集成进来和dotnetty 做下对比，dotnetty 有多强，我压测可以支持20w/s, 然后客户提供的服务器，通过外网压测网关，把上行速度50MB带宽的网络跑满了，引擎主机CPU只是在15%左右，完全没有跑满。然后再试试国人开发的SuperSocket看下性能怎么样。

 

 

木舟 (Kayak) 是什么?
       木舟(Kayak)是基于.NET6.0软件环境下的surging微服务引擎进行开发的, 平台包含了微服务和物联网平台。支持异步和响应式编程开发，功能包含了物模型,设备,产品,网络组件的统一管理和微服务平台下的注册中心，服务路由，模块，中间服务等管理。还有多协议适配(TCP,MQTT,UDP,CoAP,HTTP,Grpc,websocket,rtmp,httpflv,webservice,等),通过灵活多样的配置适配能够接入不同厂家不同协议等设备。并且通过设备告警,消息通知,数据可视化等功能。能够让你能快速建立起微服务物联网平台系统。
     凯亚物联网平台：http://117.72.121.2:3100（用户名：fanly  密码：123456）（木舟物联网有人取了，准备改名原神凯亚，凡是交托于他的任务，总能得到解决）
    链路跟踪Skywalking V8:http://117.72.121.2:8080/
      surging 微服务引擎开源地址：https://github.com/fanliang11/surging（后面surging 会移动到microsurging进行维护）
 
二、集成SuperSocket

作为去中心化的微服务引擎，相关的引擎组件，中间件都可以替换，就比如核心的RPC组件dotnetty 都可以替换成其它组件，下面介绍如何进行替换
创建服务端消息监听SuperSocketServerMessageListener，需要继承IMessageListener，代码如下：

1.   public class SuperSocketServerMessageListener : IMessageListener, IDisposable
2.   {
3.       public event ReceivedDelegate Received;
5.       private readonly ILogger<SuperSocketServerMessageListener> _logger;
7.       private readonly ITransportMessageDecoder _transportMessageDecoder;
8.       private readonly ITransportMessageEncoder _transportMessageEncoder;
9.       private readonly IServiceEngineLifetime _serviceEngineLifetime;
11.       public SuperSocketServerMessageListener(ILogger<SuperSocketServerMessageListener> logger, ITransportMessageCodecFactory codecFactory, IServiceEngineLifetime serviceEngineLifetime)
12.       {
13.           _logger = logger;
14.           _transportMessageEncoder = codecFactory.GetEncoder();
15.           _transportMessageDecoder = codecFactory.GetDecoder();
16.           _serviceEngineLifetime = serviceEngineLifetime;
17.       }
18.       public async Task StartAsync(EndPoint endPoint)
19.       {
21.           _serviceEngineLifetime.ServiceEngineStarted.Register(async () =>
22.           {
23.               try
24.               {
25.                   var ipEndPoint = endPoint as IPEndPoint;
26.                   var host = SuperSocketHostBuilder.Create<TransportMessage, TransportMessagePipelineFilter>()
27.               
28.                   .UsePackageHandler( (s, p) =>
29.                   {
30.                       Task.Run(async () =>
31.                       {
32.                           var sender = new SuperSocketServerMessageSender(_transportMessageEncoder, s);
33.                           await OnReceived(sender, p);
34.                       });
35.                       return ValueTask.CompletedTask;
36.                   })
37.                   .ConfigureSuperSocket(options =>
38.                   {
39.                       options.Name = "Echo Server";
40.                       options.Logger = _logger;
41.                       options.AddListener(new ListenOptions
42.                       {
43.                           Ip = ipEndPoint.Address.ToString(),
44.                           Port = ipEndPoint.Port,
45.                         
46.                       }
47.                       );
48.                   })
49.                   .ConfigureLogging((hostCtx, loggingBuilder) =>
50.                   {
51.                       loggingBuilder.AddConsole();
52.                   })
53.                   .Build();
54.                   await host.RunAsync();
55.               }
56.               catch (Exception ex)
57.               {
58.                   _logger.LogError($"SuperSocket服务主机启动失败，监听地址：{endPoint}。 ");
59.               }
60.           });
62.       }
63.       
64.       public async Task OnReceived(IMessageSender sender, TransportMessage message)
65.       {
66.           if (Received == null)
67.               return;
68.           await Received(sender, message);
69.       }
71.       public void Dispose()
72.       {
73.       }
74.   }

复制代码

创建客户端消息监听SuperSocketTransportClientFactory，需要继承ITransportClientFactory，代码如下：
 

1. internal class SuperSocketTransportClientFactory : ITransportClientFactory, IDisposable
2. {
3.      private readonly ITransportMessageEncoder _transportMessageEncoder;
4.      private readonly ITransportMessageDecoder _transportMessageDecoder;
5.      private readonly ILogger<SuperSocketTransportClientFactory> _logger;
6.      private readonly IServiceExecutor _serviceExecutor;
7.      private readonly IHealthCheckService _healthCheckService;
8.      private readonly ConcurrentDictionary<EndPoint, Lazy<Task<ITransportClient>>> _clients = new ConcurrentDictionary<EndPoint, Lazy<Task<ITransportClient>>>();
10.      public SuperSocketTransportClientFactory(ITransportMessageCodecFactory codecFactory, IHealthCheckService healthCheckService, ILogger<SuperSocketTransportClientFactory> logger)
11. : this(codecFactory, healthCheckService, logger, null)
12.      {
13.      }
15.      public SuperSocketTransportClientFactory(ITransportMessageCodecFactory codecFactory, IHealthCheckService healthCheckService, ILogger<SuperSocketTransportClientFactory> logger, IServiceExecutor serviceExecutor)
16.      {
17.          _transportMessageEncoder = codecFactory.GetEncoder();
18.          _transportMessageDecoder = codecFactory.GetDecoder();
19.          _logger = logger;
20.          _serviceExecutor = serviceExecutor;
21.          _healthCheckService = healthCheckService;
22.      }
23.      public async Task<ITransportClient> CreateClientAsync(EndPoint endPoint)
24.      {
25.          var key = endPoint;
26.          if (_logger.IsEnabled(LogLevel.Debug))
27.              _logger.LogDebug($"准备为服务端地址：{key}创建客户端。");
28.          try
29.          {
30.              return await _clients.GetOrAdd(key
31.           , k => new Lazy<Task<ITransportClient>>(async () =>
32.           {
33.               //客户端对象
34.               var client = new EasyClient<TransportMessage>(new TransportMessagePipelineFilter()).AsClient();
35.               var messageListener = new MessageListener();
36.               var messageSender = new SuperSocketMessageClientSender(_transportMessageEncoder, client);
37.               await client.ConnectAsync(endPoint);
38.               client.PackageHandler += async (sender, package) =>
39.               {
40.                   await messageListener.OnReceived(messageSender, package);
41.               };
42.               client.StartReceive();
43.               //创建客户端
44.               var transportClient = new TransportClient(messageSender, messageListener, _logger, _serviceExecutor);
45.               return transportClient;
46.           }
47.               )).Value;//返回实例
48.          }
49.          catch
50.          {
51.              //移除
52.              _clients.TryRemove(key, out var value);
53.              var ipEndPoint = endPoint as IPEndPoint;
54.              //标记这个地址是失败的请求
55.              if (ipEndPoint != null)
56.                  await _healthCheckService.MarkFailure(new IpAddressModel(ipEndPoint.Address.ToString(), ipEndPoint.Port));
57.              throw;
58.          }
59.      }
62.      public void Dispose()
63.      {
64.          foreach (var client in _clients.Values)
65.          {
66.              (client as IDisposable)?.Dispose();
67.          }
68.      }
69. }

复制代码

注册初始化SuperSocket引擎模块，需要继承EnginePartModule， 代码如下：

1.   public class SuperSocketModule : EnginePartModule
2.   {
3.       public override void Initialize(AppModuleContext context)
4.       {
5.           base.Initialize(context);
6.       }
8.       /// <summary>
9.       /// Inject dependent third-party components
10.       /// </summary>
11.       /// <param name="builder"></param>
12.       protected override void RegisterBuilder(ContainerBuilderWrapper builder)
13.       {
14.           base.RegisterBuilder(builder);
15.           builder.Register(provider =>
16.           {
17.               IServiceExecutor serviceExecutor = null;
18.               if (provider.IsRegistered(typeof(IServiceExecutor)))
19.                   serviceExecutor = provider.Resolve<IServiceExecutor>();
20.               return new SuperSocketTransportClientFactory(provider.Resolve<ITransportMessageCodecFactory>(),
21.                     provider.Resolve<IHealthCheckService>(),
22.                   provider.Resolve<ILogger<SuperSocketTransportClientFactory>>(),
23.                   serviceExecutor);
24.           }).As(typeof(ITransportClientFactory)).SingleInstance();
25.           if (AppConfig.ServerOptions.Protocol == CommunicationProtocol.Tcp ||
26.               AppConfig.ServerOptions.Protocol == CommunicationProtocol.None)
27.           {
28.               RegisterDefaultProtocol(builder);
29.           }
30.       }
32.       private void RegisterDefaultProtocol(ContainerBuilderWrapper builder)
33.       {
34.           builder.Register(provider =>
35.           {
36.               return new SuperSocketServerMessageListener(provider.Resolve<ILogger<SuperSocketServerMessageListener>>(),
37.                     provider.Resolve<ITransportMessageCodecFactory>(),
38.                        provider.Resolve<IServiceEngineLifetime>());
39.           }).SingleInstance();
40.           builder.Register(provider =>
41.           {
42.               var serviceExecutor = provider.ResolveKeyed<IServiceExecutor>(CommunicationProtocol.Tcp.ToString());
43.               var messageListener = provider.Resolve<SuperSocketServerMessageListener>();
44.               return new DefaultServiceHost(async endPoint =>
45.               {
46.                   await messageListener.StartAsync(endPoint);
47.                   return messageListener;
48.               }, serviceExecutor);
49.           }).As<IServiceHost>();
50.       }
51.   }

复制代码

客户端服务端消息发送，需要继承IMessageSender， 代码如下：
 

1.     public abstract class SuperSocketMessageSender
2.     {
3.         private readonly ITransportMessageEncoder _transportMessageEncoder;
5.         protected SuperSocketMessageSender(ITransportMessageEncoder transportMessageEncoder)
6.         {
7.             _transportMessageEncoder = transportMessageEncoder;
8.         }
10.         protected byte[] GetByteBuffer(TransportMessage message)
11.         {
12.             var data = _transportMessageEncoder.Encode(message).ToList();
13.             data.AddRange(Encoding.UTF8.GetBytes("\r\n"));
14.             //var buffer = PooledByteBufferAllocator.Default.Buffer();
15.             return data.ToArray();
16.         }
17.     }
19.     public class SuperSocketMessageClientSender : SuperSocketMessageSender, IMessageSender
20.     {
21.         private readonly IEasyClient<TransportMessage> _client;
23.         public SuperSocketMessageClientSender(ITransportMessageEncoder transportMessageEncoder, IEasyClient<TransportMessage> client) : base(transportMessageEncoder)
24.         {
25.             _client = client;
26.         }
28.         /// <summary>
29.         /// 发送消息。
30.         /// </summary>
31.         /// <param name="message">消息内容。</param>
32.         /// <returns>一个任务。</returns>
33.         public async Task SendAsync(TransportMessage message)
34.         {
35.             var buffer = GetByteBuffer(message);
36.            await _client.SendAsync(buffer);
37.         }
39.         /// <summary>
40.         /// 发送消息并清空缓冲区。
41.         /// </summary>
42.         /// <param name="message">消息内容。</param>
43.         /// <returns>一个任务。</returns>
44.         public async Task SendAndFlushAsync(TransportMessage message)
45.         {
46.             var buffer = GetByteBuffer(message);
47.             await _client.SendAsync(buffer);
48.             // _client.StartReceive();
49.             //var p=  await _client.ReceiveAsync();
50.         }
51.     }
53.     #region Implementation of IMessageSender
54.     public class SuperSocketServerMessageSender : SuperSocketMessageSender, IMessageSender
55.     {
56.         private readonly IAppSession _session;
58.         public SuperSocketServerMessageSender(ITransportMessageEncoder transportMessageEncoder, IAppSession session) : base(transportMessageEncoder)
59.         {
60.             _session = session;
61.         }
64.         /// <summary>
65.         /// 发送消息。
66.         /// </summary>
67.         /// <param name="message">消息内容。</param>
68.         /// <returns>一个任务。</returns>
69.         public async Task SendAsync(TransportMessage message)
70.         {
71.             var buffer = GetByteBuffer(message);
72.            await _session.SendAsync(buffer);
73.         }
75.         /// <summary>
76.         /// 发送消息并清空缓冲区。
77.         /// </summary>
78.         /// <param name="message">消息内容。</param>
79.         /// <returns>一个任务。</returns>
80.         public async Task SendAndFlushAsync(TransportMessage message)
81.         {
82.             var buffer = GetByteBuffer(message);
83.            await _session.SendAsync(buffer);
84.         }
86.     }
87.     #endregion

复制代码

 
SuperSocket过滤器，需要继承TerminatorPipelineFilter， 代码如下：
 

1.     public class TransportMessagePipelineFilter : TerminatorPipelineFilter<TransportMessage>
2.     {
3.         private readonly ITransportMessageDecoder _transportMessageDecoder;
4.         public TransportMessagePipelineFilter() : base(new[] { (byte)'\r', (byte)'\n' })
5.         {
6.             _transportMessageDecoder = ServiceLocator.GetService<ITransportMessageCodecFactory>().GetDecoder();
7.         }
10.         public override TransportMessage Filter(ref SequenceReader<byte> bufferStream)
11.         {
12.             try
13.             {
14.                 var bytes = bufferStream.Sequence.Slice(0, bufferStream.Length - 2).ToArray();
15.                 var transportMessage = _transportMessageDecoder.Decode(bytes);
16.                 return transportMessage;
17.             }
18.             finally
19.             {
20.                 bufferStream.Advance(bufferStream.Length);
21.             }
22.         }
23.     }

复制代码

 
三、如何加载SuperSocket引擎组件

第一种方式：
去掉Surging.Core.DotNetty引用，添加Surging.Core.SuperSocket

 第二种方式
添加Surging.Core.DotNetty，Surging.Core.SuperSocket这两个应用，在surgingSettings.json配置文件中把Packages的using列表中的DotNettyModule改成SuperSocketModule

四、结果

可能是预发布版本，在测试当中还是有些问题，kayka物联网平台换成SuperSocket还是会发生错误，暂时没有条件进行压测，可能是因为预发布版本，作者还需要完善，等正式版之后再压测做对比吧
 
五、总结

因为这段时间比较忙，还需要协助客户拆分服务，缓存降级，消息队列，等忙完这段时间，线上物联网平台会开通端口给用户进行测试，我也会努力把物联网进行完善，让微服务物联网平台能走向新的高度。
 

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