经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 程序设计 » ASP.net » 查看文章
基于DotNetty实现自动发布 - 通信实现
来源：cnblogs　　作者：Broadm　　时间：2023/12/5 12:10:55　　对本文有异议

基于 DotNetty 实现通信

DotNetty : 是微软的 Azure 团队，使用 C#实现的 Netty 的版本发布。是.NET 平台的优秀网络库。

项目介绍

OpenDeploy.Communication 类库项目,是通信相关基础设施层

image

  • Codec 模块实现编码解码
  • Convention 模块定义约定,比如抽象的业务 Handler, 消息载体 NettyMessage, 消息上下文 'NettyContext' 等

自定义消息格式

消息类为 NettyMessage ,封装了消息头 NettyHeader 和消息体 Body

image

NettyMessage

封装了消息头 NettyHeader 和消息体 Body

NettyMessage 点击查看代码 
  1. /// <summary> Netty消息 </summary>
  2. public class NettyMessage
  3. {
  4.     /// <summary> 消息头 </summary>
  5.     public NettyHeader Header { get; init; } = default!;
  7.     /// <summary> 消息体(可空,可根据具体业务而定) </summary>
  8.     public byte[]? Body { get; init; }
  10.     /// <summary> 消息头转为字节数组 </summary>
  11.     public byte[] GetHeaderBytes()
  12.     {
  13.         var headerString = Header.ToString();
  14.         return Encoding.UTF8.GetBytes(headerString);
  15.     }
  17.     /// <summary> 是否同步消息 </summary>
  18.     public bool IsSync() => Header.Sync;
  20.     /// <summary> 创建Netty消息工厂方法 </summary>
  21.     public static NettyMessage Create(string endpoint, bool sync = false, byte[]? body = null)
  22.     {
  23.         return new NettyMessage
  24.         {
  25.             Header = new NettyHeader { EndPoint = endpoint, Sync = sync },
  26.             Body = body
  27.         };
  28.     }
  30.     /// <summary> 序列化为JSON字符串 </summary>
  31.     public override string ToString() => Header.ToString();
  32. }

NettyHeader

消息头,包含请求唯一标识,是否同步消息,终结点等, 在传输数据时会序列化为 JSON

NettyHeader 点击查看代码 
  1. /// <summary> Netty消息头 </summary>
  2. public class NettyHeader
  3. {
  4.     /// <summary> 请求消息唯一标识 </summary>
  5.     public Guid RequestId { get; init; } = Guid.NewGuid();
  7.     /// <summary> 是否同步消息, 默认false是异步消息 </summary>
  8.     public bool Sync { get; init; }
  10.     /// <summary> 终结点 (借鉴MVC,约定为Control/Action模式) </summary>
  11.     public string EndPoint { get; init; } = string.Empty;
  13.     /// <summary> 序列化为JSON字符串 </summary>
  14.     public override string ToString() => this.ToJsonString();
  15. }
  • 请求消息唯一标识 RequestId , 用来唯一标识消息, 主要用于 发送同步请求, 因为默认的消息是异步的,只管发出去,不需要等待响应
  • 是否同步消息 Sync , 可以不需要,主要为了可视化,便于调试
  • 终结点 EndPoint , (借鉴 MVC,约定为 Control/Action 模式), 服务端直接解析出对应的处理器

编码器

DefaultEncoder 点击查看代码 
  1. public class DefaultEncoder : MessageToByteEncoder<NettyMessage>
  2. {
  3.     protected override void Encode(IChannelHandlerContext context, NettyMessage message, IByteBuffer output)
  4.     {
  5.         //消息头转为字节数组
  6.         var headerBytes = message.GetHeaderBytes();
  8.         //写入消息头长度
  9.         output.WriteInt(headerBytes.Length);
  11.         //写入消息头字节数组
  12.         output.WriteBytes(headerBytes);
  14.         //写入消息体字节数组
  15.         if (message.Body != null && message.Body.Length > 0)
  16.         {
  17.             output.WriteBytes(message.Body);
  18.         }
  19.     }
  20. }

解码器

DefaultDecoder 点击查看代码 
  1. public class DefaultDecoder : MessageToMessageDecoder<IByteBuffer>
  2. {
  3.     protected override void Decode(IChannelHandlerContext context, IByteBuffer input, List<object> output)
  4.     {
  5.         //消息总长度
  6.         var totalLength = input.ReadableBytes;
  8.         //消息头长度
  9.         var headerLength = input.GetInt(input.ReaderIndex);
  11.         //消息体长度
  12.         var bodyLength = totalLength - 4 - headerLength;
  14.         //读取消息头字节数组
  15.         var headerBytes = new byte[headerLength];
  16.         input.GetBytes(input.ReaderIndex + 4, headerBytes, 0, headerLength);
  18.         byte[]? bodyBytes = null;
  19.         string? rawHeaderString = null;
  20.         NettyHeader? header;
  22.         try
  23.         {
  24.             //把消息头字节数组,反序列化为JSON
  25.             rawHeaderString = Encoding.UTF8.GetString(headerBytes);
  26.             header = JsonSerializer.Deserialize<NettyHeader>(rawHeaderString);
  27.         }
  28.         catch (Exception ex)
  29.         {
  30.             Logger.Error($"解码失败: {rawHeaderString}, {ex}");
  31.             return;
  32.         }
  34.         if (header is null)
  35.         {
  36.             Logger.Error($"解码失败: {rawHeaderString}");
  37.             return;
  38.         }
  40.         //读取消息体字节数组
  41.         if (bodyLength > 0)
  42.         {
  43.             bodyBytes = new byte[bodyLength];
  44.             input.GetBytes(input.ReaderIndex + 4 + headerLength, bodyBytes, 0, bodyLength);
  45.         }
  47.         //封装为NettyMessage对象
  48.         var message = new NettyMessage
  49.         {
  50.             Header = header,
  51.             Body = bodyBytes,
  52.         };
  54.         output.Add(message);
  55.     }
  56. }

NettyServer 实现

NettyServer 点击查看代码 
  1. public static class NettyServer
  2. {
  3.     /// <summary>
  4.     /// 开启Netty服务
  5.     /// </summary>
  6.     public static async Task RunAsync(int port = 20007)
  7.     {
  8.         var bossGroup = new MultithreadEventLoopGroup(1);
  9.         var workerGroup = new MultithreadEventLoopGroup();
  11.         try
  12.         {
  13.             var bootstrap = new ServerBootstrap().Group(bossGroup, workerGroup);
  15.             bootstrap
  16.                 .Channel<TcpServerSocketChannel>()
  17.                 .Option(ChannelOption.SoBacklog, 100)
  18.                 .Option(ChannelOption.SoReuseaddr, true)
  19.                 .Option(ChannelOption.SoReuseport, true)
  20.                 .ChildHandler(new ActionChannelInitializer<IChannel>(channel =>
  21.                 {
  22.                     IChannelPipeline pipeline = channel.Pipeline;
  23.                     pipeline.AddLast("framing-enc", new LengthFieldPrepender(4));
  24.                     pipeline.AddLast("framing-dec", new LengthFieldBasedFrameDecoder(int.MaxValue, 0, 4, 0, 4));
  25.                     pipeline.AddLast("decoder", new DefaultDecoder());
  26.                     pipeline.AddLast("encoder", new DefaultEncoder());
  27.                     pipeline.AddLast("handler", new ServerMessageEntry());
  28.                 }));
  30.             var boundChannel = await bootstrap.BindAsync(port);
  32.             Logger.Info($"NettyServer启动成功...{boundChannel}");
  34.             Console.ReadLine();
  36.             await boundChannel.CloseAsync();
  38.             Logger.Info($"NettyServer关闭监听了...{boundChannel}");
  39.         }
  40.         finally
  41.         {
  42.             await Task.WhenAll(
  43.                 bossGroup.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1)),
  44.                 workerGroup.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1))
  45.             );
  47.             Logger.Info($"NettyServer退出了...");
  48.         }
  50.     }
  51. }
  • 服务端管道最后我们添加了 ServerMessageEntry ,作为消息处理的入口
ServerMessageEntry 点击查看代码 
  1. public class ServerMessageEntry : ChannelHandlerAdapter
  2. {
  3.     /// <summary> Netty处理器选择器 </summary>
  4.     private readonly DefaultNettyHandlerSelector handlerSelector = new();
  6.     public ServerMessageEntry()
  7.     {
  8.         //注册Netty处理器
  9.         handlerSelector.RegisterHandlerTypes(typeof(EchoHandler).Assembly.GetTypes());
  10.     }
  12.     /// <summary> 通道激活 </summary>
  13.     public override void ChannelActive(IChannelHandlerContext context)
  14.     {
  15.         Logger.Warn($"ChannelActive: {context.Channel}");
  16.     }
  18.     /// <summary> 通道关闭 </summary>
  19.     public override void ChannelInactive(IChannelHandlerContext context)
  20.     {
  21.         Logger.Warn($"ChannelInactive: {context.Channel}");
  22.     }
  24.     /// <summary> 收到客户端的消息 </summary>
  25.     public override async void ChannelRead(IChannelHandlerContext context, object message)
  26.     {
  27.         if (message is not NettyMessage nettyMessage)
  28.         {
  29.             Logger.Error("从客户端接收消息为空");
  30.             return;
  31.         }
  33.         try
  34.         {
  35.             Logger.Info($"收到客户端的消息: {nettyMessage}");
  37.             //封装请求
  38.             var nettyContext = new NettyContext(context.Channel, nettyMessage);
  40.             //选择处理器
  41.             AbstractNettyHandler handler = handlerSelector.SelectHandler(nettyContext);
  43.             //处理请求
  44.             await handler.ProcessAsync();
  45.         }
  46.         catch(Exception ex)
  47.         {
  48.             Logger.Error($"ServerMessageEntry.ChannelRead: {ex}");
  49.         }
  50.     }
  51. }

  • 按照约定, 把继承 AbstractNettyHandler 的类视为业务处理器

  • ServerMessageEntry 拿到消息后,首先把消息封装为 NettyContext, 类似与 MVC 中的 HttpContext, 封装了请求和响应对象, 内部解析请求的 EndPoint, 拆分为 HandlerName, ActionName

  • DefaultNettyHandlerSelector 提供注册处理器的方法 RegisterHandlerTypes, 和选择处理器的方法 SelectHandler

  • SelectHandler, 默认规则是查找已注册的处理器中以 HandlerName 开头的类型

  • AbstractNettyHandler 的 ProcessAsync 方法,通过 ActionName, 反射拿到 MethodInfo, 调用终结点

NettyClient 实现

NettyClient 点击查看代码 
  1. public sealed class NettyClient(string serverHost, int serverPort) : IDisposable
  2. {
  3.     public EndPoint ServerEndPoint { get; } = new IPEndPoint(IPAddress.Parse(serverHost), serverPort);
  5.     private static readonly Bootstrap bootstrap = new();
  6.     private static readonly IEventLoopGroup eventLoopGroup = new SingleThreadEventLoop();
  8.     private bool _disposed;
  9.     private IChannel? _channel;
  10.     public bool IsConnected => _channel != null && _channel.Open;
  11.     public bool IsWritable => _channel != null && _channel.IsWritable;
  13.     static NettyClient()
  14.     {
  15.         bootstrap
  16.             .Group(eventLoopGroup)
  17.             .Channel<TcpSocketChannel>()
  18.             .Option(ChannelOption.SoReuseaddr, true)
  19.             .Option(ChannelOption.SoReuseport, true)
  20.             .Handler(new ActionChannelInitializer<ISocketChannel>(channel =>
  21.             {
  22.                 IChannelPipeline pipeline = channel.Pipeline;
  23.                 //pipeline.AddLast("ping", new IdleStateHandler(0, 5, 0));
  24.                 pipeline.AddLast("framing-enc", new LengthFieldPrepender(4));
  25.                 pipeline.AddLast("framing-dec", new LengthFieldBasedFrameDecoder(int.MaxValue, 0, 4, 0, 4));
  26.                 pipeline.AddLast("decoder", new DefaultDecoder());
  27.                 pipeline.AddLast("encoder", new DefaultEncoder());
  28.                 pipeline.AddLast("handler", new ClientMessageEntry());
  29.             }));
  30.     }
  32.     /// <summary> 连接服务器 </summary>
  33.     private async Task TryConnectAsync()
  34.     {
  35.         try
  36.         {
  37.             if (IsConnected) { return; }
  38.             _channel = await bootstrap.ConnectAsync(ServerEndPoint);
  39.         }
  40.         catch (Exception ex)
  41.         {
  42.             throw new Exception($"连接服务器失败 : {ServerEndPoint} {ex.Message}");
  43.         }
  44.     }
  46.     /// <summary>
  47.     /// 发送消息
  48.     /// </summary>
  49.     /// <param name="endpoint">终结点</param>
  50.     /// <param name="sync">是否同步等待响应</param>
  51.     /// <param name="body">正文</param>
  52.     public async Task SendAsync(string endpoint, bool sync = false, byte[]? body = null)
  53.     {
  54.         var message = NettyMessage.Create(endpoint, sync, body);
  55.         if (sync)
  56.         {
  57.             var task = ClientMessageSynchronizer.TryAdd(message);
  58.             try
  59.             {
  60.                 await SendAsync(message);
  61.                 await task;
  62.             }
  63.             catch
  64.             {
  65.                 ClientMessageSynchronizer.TryRemove(message);
  66.                 throw;
  67.             }
  68.         }
  69.         else
  70.         {
  71.             await SendAsync(message);
  72.         }
  73.     }
  75.     /// <summary>
  76.     /// 发送消息
  77.     /// </summary>
  78.     private async Task SendAsync(NettyMessage message)
  79.     {
  80.         await TryConnectAsync();
  81.         await _channel!.WriteAndFlushAsync(message);
  82.     }
  84.     /// <summary> 释放连接(程序员手动释放, 一般在代码使用using语句,或在finally里面Dispose) </summary>
  85.     public void Dispose()
  86.     {
  87.         Dispose(true);
  88.         GC.SuppressFinalize(this);
  89.     }
  91.     /// <summary> 释放连接 </summary>
  92.     private void Dispose(bool disposing)
  93.     {
  94.         if (_disposed)
  95.         {
  96.             return;
  97.         }
  99.         //释放托管资源,比如嵌套的对象
  100.         if (disposing)
  101.         {
  103.         }
  105.         //释放非托管资源
  106.         if (_channel != null)
  107.         {
  108.             _channel.CloseAsync();
  109.             _channel = null;
  110.         }
  112.         _disposed = true;
  113.     }
  115.     ~NettyClient()
  116.     {
  117.         Dispose(true);
  118.     }
  119. }
  • NettyClient 封装了 Netty 客户端逻辑,提供发送异步请求(默认)和发布同步请求方法
  • DotNetty 默认不提供同步请求,但是有些情况我们需要同步等待服务器的响应,所有需要自行实现,实现也很简单,把消息 ID 缓存起来,收到服务器响应后激活就行了,具体实现在消息同步器 ClientMessageSynchronizer, 就不贴了

总结

至此,我们实现了基于 DotNetty 搭建通信模块, 实现了客户端和服务器的编解码,处理器选择,客户端实现了同步消息等,大家可以在 ConsoleHost 结尾的控制台项目中,测试下同步和异步的消息,实现的简单的 Echo 模式

代码仓库

项目暂且就叫 OpenDeploy 吧

欢迎大家拍砖,Star

下一步

计划下一步,基于WPF的客户端, 实现接口项目的配置与发现

原文链接：https://www.cnblogs.com/broadm/p/17875559.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

ASP.net热门文章

  • 【Dotnet 工具箱】WPF UI - 现代化设计的开源 WPF 框架
  • net core 使用 SqlSugar
  • .NETCore下CI/CD之自动化测试
  • .net core部署到linux可能碰到的问题
  • Abp vNext 自定义 Ef Core 仓储引发异常
  • .NetCore 消息队列的使用
  • .NetCore WebApi——基于JWT的简单身份认证与授权（Swagger）
  • Abp.NHibernate连接PostgreSQl数据库
  • ModelValidator基于元数据的验证
  • .NET MAUI 安卓应用开发初体验

    • ASP.net推荐文章

  • Gradio.NET 支持 .NET 8 简化 Web 应用开发
  • .NET 轻量化定时任务调度 FreeScheduler
  • WPF 模仿前端大佬写一个Hover效果
  • WPF 模拟UWP原生窗口样式——亚克力|云母材质、自定义标题栏样式、原生DWM动画 （附我封装好的类）
  • .NET 音频采集
  • .NET8 Blazor 从入门到精通：（三）类库和表单
  • wpf 如何7步写一个badge控件
  • .NET 高效Nuget管理工具（开源）
  • .NET 8 跨平台高性能边缘采集网关
  • .NET 窗口/屏幕录制

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号