.NET Core3.1 Dotnetty实战第一章

时间：2020-05-17 17:36:29 阅读：103 评论：0 收藏：0 [点我收藏+]

一、概要

本系列文章主要讲述由微软Azure团队研发的.net的版本的netty，Dotnetty。所有的开发都将基于.net core 3.1版本进行开发。

Dotnetty是什么，原本Netty是由JBOSS提供的一个java开源框架后来由微软抄了一份.net的版本, 是业界最流行的NIO框架，整合了多种协议（包括FTP、SMTP、 HTTP等各种二进制文本协议）的实现经验，精心设计的框架，在多个大型商业项目中得到充分验证。

个人使用感受如下：

1.Dotnetty各方面封装的很好，不要开发者过度关系细节。除了消息协议处理方面（socket网络通信的分包粘包处理）。

2.使用非常便捷，语法、和各组件结构清晰可重用性高。

3.也是.net core中为数不多看起来比较靠谱的框架，为什么会这么说呢在做股票相关项目时需求调研和技术选型的时候看了很多网络通信框架。

要么就是.Net Framework的版本，就怕有些开源团队不持续更新导致致命bug解决成本高等等问题。

二、简介

本篇文章主要围绕dotnetty基础概念和相关知识点来讲：

1. NIO和BIO的概念

2. 相关网络知识

socket交互流程，字节序和网络字节序

3. Dotnetty 框架介绍

4. Dotnetty Demo的讲解

三、主要内容

NIO和BIO、AIO的概念（摘抄自：https://zhuanlan.zhihu.com/p/111816019）

BIO（同步阻塞）：客户端在请求数据的过程中，保持一个连接，不能做其他事情。

技术图片

NIO（同步非阻塞）：客户端在请求数据的过程中，不用保持一个连接，不能做其他事情。（不用保持一个连接，而是用许多个小连接，也就是轮询）

技术图片

AIO（异步非阻塞）：客户端在请求数据的过程中，不用保持一个连接，可以做其他事情。（客户端做其他事情，数据来了等服务端来通知。）

Dotnetty 框架介绍

技术图片

目前个人使用下来，主要用到的核心内容如上。

DotNetty.Common 是公共的类库项目，包装线程池，并行任务和常用帮助类的封装
DotNetty.Transport 是DotNetty核心的实现例如：Bootstrapping程序引导类，Channels 管道类（socket每有一个连接客户端就会创建一个channne）等等
DotNetty.Buffers 是对内存缓冲区管理的封装（主要在接收和发出是对socket通讯内容进行缓存管理）
DotNetty.Codes 是对编码器解码器的封装，包括一些基础基类的实现，我们在项目中自定义的协议，都要继承该项目的特定基类和实现（该类库在整个通讯环节是重中之重）
DotNetty.Handlers 封装了常用的管道处理器，比如Tls编解码，超时机制，心跳检查，日志等。（企业级开发中必不可少的处理类）

Dotnetty Demo的讲解

源码及演示代码都在官方github上： https://github.com/Azure/DotNetty

开发参考文档：https://netty.io/wiki/index.html （开发文档是java的版本，dotnetty都是对着java抄的会有不一样的地方但是大部分都相同。目前没有看到比较权威.net版本的文档）

技术图片

下面主要分为两个部分去讲解：

Server部分

// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

namespace Echo.Server
{
using System;
using System.IO;
using System.Security.Cryptography.X509Certificates;
using System.Threading.Tasks;
using DotNetty.Codecs;
using DotNetty.Handlers.Logging;
using DotNetty.Handlers.Tls;
using DotNetty.Transport.Bootstrapping;
using DotNetty.Transport.Channels;
using DotNetty.Transport.Channels.Sockets;
using DotNetty.Transport.Libuv;
using Examples.Common;

class Program
{
static async Task RunServerAsync()
{
ExampleHelper.SetConsoleLogger();

IEventLoopGroup bossGroup;//主要工作组，设置为2个线程
IEventLoopGroup workerGroup;//子工作组，推荐设置为内核数*2的线程数

if (ServerSettings.UseLibuv)
{
var dispatcher = new DispatcherEventLoopGroup();
bossGroup = dispatcher;
workerGroup = new WorkerEventLoopGroup(dispatcher);
}
else
{
bossGroup = new MultithreadEventLoopGroup(1);//主线程只会实例化一个
workerGroup = new MultithreadEventLoopGroup();//子线程组可以按照自己的需求在构造函数里指定数量
}

X509Certificate2 tlsCertificate = null;
if (ServerSettings.IsSsl)//是否使用ssl套接字加密
{
tlsCertificate = new X509Certificate2(Path.Combine(ExampleHelper.ProcessDirectory, "dotnetty.com.pfx"), "password");
}
try
{
/*
*ServerBootstrap是一个引导类，表示实例化的是一个服务端对象
*声明一个服务端Bootstrap，每个Netty服务端程序，都由ServerBootstrap控制，
*通过链式的方式组装需要的参数
*/
var bootstrap = new ServerBootstrap();
//添加工作组，其中内部实现为将子线程组内置到主线程组中进行管理
bootstrap.Group(bossGroup, workerGroup);

if (ServerSettings.UseLibuv)//这个ifelse中实例化的是工作频道，就是处理读取或者发送socket数据的地方
{
bootstrap.Channel<TcpServerChannel>();
}
else
{
bootstrap.Channel<TcpServerSocketChannel>();
}

bootstrap
.Option(ChannelOption.SoBacklog, 100)
.Option(ChannelOption.SoReuseport, true)//设置端口复用
.Handler(new LoggingHandler("SRV-LSTN"))//初始化日志拦截器
.ChildHandler(new ActionChannelInitializer<IChannel>(channel =>//初始化Tcp服务
{
/*
* 这里主要是配置channel中需要被设置哪些参数，以及channel具体的实现方法内容。
* channel可以理解为，socket通讯当中客户端和服务端的连接会话，会话内容的处理在channel中实现。
*/

IChannelPipeline pipeline = channel.Pipeline;
if (tlsCertificate != null)
{
pipeline.AddLast("tls", TlsHandler.Server(tlsCertificate));//添加ssl加密
}
pipeline.AddLast(new LoggingHandler("SRV-CONN"));
pipeline.AddLast("framing-enc", new LengthFieldPrepender(2));//Dotnetty自带的编码器，将要发送的内容进行编码然后发送
pipeline.AddLast("framing-dec", new LengthFieldBasedFrameDecoder(ushort.MaxValue, 0, 2, 0, 2));//Dotnetty自带的解码器，将接受到的内容进行解码然后根据内容对应到业务逻辑当中

pipeline.AddLast("echo", new EchoServerHandler());//server的channel的处理类实现

}));

IChannel boundChannel = await bootstrap.BindAsync(ServerSettings.Port);//指定服务端的端口号，ip地址donetty可以自动获取到本机的地址。也可以在这里手动指定。

Console.ReadLine();

await boundChannel.CloseAsync();//关闭
}
finally
{
//关闭释放并退出
await Task.WhenAll(
bossGroup.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1)),
workerGroup.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1)));
}
}

static void Main() => RunServerAsync().Wait();
}
}

channel的实现细节（socket会话内容处理和业务逻辑都可以在这里处理）

// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

namespace Echo.Server
{
    using System;
    using System.Text;
    using DotNetty.Buffers;
    using DotNetty.Transport.Channels;

    /// <summary>
    /// 该类为Server的Channel具体和定义实现
    /// </summary>
    public class EchoServerHandler : ChannelHandlerAdapter
    {
        /*
         * Channel的生命周期
         * 1.ChannelRegistered 先注册
         * 2.ChannelActive 再被激活
         * 3.ChannelRead 客户端与服务端建立连接之后的会话（数据交互）
         * 4.ChannelReadComplete 读取客户端发送的消息完成之后
         * error. ExceptionCaught 如果在会话过程当中出现dotnetty框架内部异常都会通过Caught方法返回给开发者
         * 5.ChannelInactive 使当前频道处于未激活状态
         * 6.ChannelUnregistered 取消注册
         */

        /// <summary>
        /// 频道注册
        /// </summary>
        /// <param name="context"></param>
        public override void ChannelRegistered(IChannelHandlerContext context)
        {
            base.ChannelRegistered(context);
        }

        /// <summary>
        /// socket client 连接到服务端的时候channel被激活的回调函数
        /// </summary>
        /// <param name="context"></param>
        public override void ChannelActive(IChannelHandlerContext context)
        {
            //一般可用来记录连接对象信息
            base.ChannelActive(context);
        }

        /// <summary>
        /// socket接收消息方法具体的实现
        /// </summary>
        /// <param name="context">当前频道的句柄，可使用发送和接收方法</param>
        /// <param name="message">接收到的客户端发送的内容</param>
        public override void ChannelRead(IChannelHandlerContext context, object message)
        {
            var buffer = message as IByteBuffer;
            if (buffer != null)
            {
                Console.WriteLine("Received from client: " + buffer.ToString(Encoding.UTF8));
            }
            context.WriteAsync(message);//这里官方的例子是直接讲客户端发送的内容原样返回给客户端，WriteAsync（）是讲要发送的内容写入到数据流的缓存中。如果不想进入数据流可以直接调用WirteAndFlusAsync（）写好了直接发送
        }

        /// <summary>
        /// 该次会话读取完成后回调函数
        /// </summary>
        /// <param name="context"></param>
        public override void ChannelReadComplete(IChannelHandlerContext context) => context.Flush();//将WriteAsync写入的数据流缓存发送出去

        /// <summary>
        /// 异常捕获
        /// </summary>
        /// <param name="context"></param>
        /// <param name="exception"></param>
        public override void ExceptionCaught(IChannelHandlerContext context, Exception exception)
        {
            Console.WriteLine("Exception: " + exception);
            context.CloseAsync();
        }

        /// <summary>
        /// 当前频道未激活状态
        /// </summary>
        /// <param name="context"></param>
        public override void ChannelInactive(IChannelHandlerContext context)
        {
            base.ChannelInactive(context);
        }

        /// <summary>
        /// 取消注册当前频道，可理解为销毁当前频道
        /// </summary>
        /// <param name="context"></param>
        public override void ChannelUnregistered(IChannelHandlerContext context)
        {
            base.ChannelUnregistered(context);
        }
    }
}

Client部分：

其他未注释部分与服务端的解释一样

// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

namespace Echo.Client
{
    using System;
    using System.IO;
    using System.Net;
    using System.Net.Security;
    using System.Security.Cryptography.X509Certificates;
    using System.Threading.Tasks;
    using DotNetty.Codecs;
    using DotNetty.Handlers.Logging;
    using DotNetty.Handlers.Tls;
    using DotNetty.Transport.Bootstrapping;
    using DotNetty.Transport.Channels;
    using DotNetty.Transport.Channels.Sockets;
    using Examples.Common;

    class Program
    {
        static async Task RunClientAsync()
        {
            ExampleHelper.SetConsoleLogger();

            var group = new MultithreadEventLoopGroup();//客户端与服务端不同的是，只需要一个主工作组进行工作协调即可不需要创建子线程组

            X509Certificate2 cert = null;
            string targetHost = null;
            if (ClientSettings.IsSsl)
            {
                cert = new X509Certificate2(Path.Combine(ExampleHelper.ProcessDirectory, "dotnetty.com.pfx"), "password");
                targetHost = cert.GetNameInfo(X509NameType.DnsName, false);
            }
            try
            {
                var bootstrap = new Bootstrap();
                bootstrap
                    .Group(group)
                    .Channel<TcpSocketChannel>()
                    .Option(ChannelOption.TcpNodelay, true)//设置为true的话不允许延迟直接发出，因为dotnetty内部实现中会将消息积累到一定的字节之后才发出。
                    .Handler(new ActionChannelInitializer<ISocketChannel>(channel =>
                    {
                        IChannelPipeline pipeline = channel.Pipeline;

                        if (cert != null)
                        {
                            pipeline.AddLast("tls", new TlsHandler(stream => new SslStream(stream, true, (sender, certificate, chain, errors) => true), new ClientTlsSettings(targetHost)));
                        }
                        pipeline.AddLast(new LoggingHandler());//donetty框架内部日志
                        pipeline.AddLast("framing-enc", new LengthFieldPrepender(2));
                        pipeline.AddLast("framing-dec", new LengthFieldBasedFrameDecoder(ushort.MaxValue, 0, 2, 0, 2));

                        pipeline.AddLast("echo", new EchoClientHandler());//client的channel的处理类实现
                    }));

                IChannel clientChannel = await bootstrap.ConnectAsync(new IPEndPoint(ClientSettings.Host, ClientSettings.Port));//设置服务端的端口号和ip地址

                Console.ReadLine();

                await clientChannel.CloseAsync();
            }
            finally
            {
                await group.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1));
            }
        }

        static void Main() => RunClientAsync().Wait();
    }
}