.NET Core微服務之路：利用DotNetty實現一個簡單的通訊過程

1 /*
2 * Netty 是一個半成品，作用是在需要基於自定義協議的基礎上完成自己的通訊封裝
3 * Netty 大大簡化了網路程式的開發過程比如 TCP 和 UDP 的 socket 服務的開發。
4 * “快速和簡單”並不意味著應用程式會有難維護和效能低的問題，
5 * Netty 是一個精心設計的框架，它從許多協議的實現中吸收了很多的經驗比如 FTP、SMTP、HTTP、許多二進位制和基於文字的傳統協議。
6 * 因此，Netty 已經成功地找到一個方式,在不失靈活性的前提下來實現開發的簡易性，高效能，穩定性。
7 */
8 
9 namespace Echo.Server
 10 {
 11using System;
 12using System.Threading.Tasks;
 13using DotNetty.Codecs;
 14using DotNetty.Handlers.Logging;
 15using DotNetty.Transport.Bootstrapping;
 16using DotNetty.Transport.Channels;
 17using DotNetty.Transport.Libuv;
 18using Examples.Common;
 19 
 20static class Program
 21{
 22static async Task RunServerAsync()
 23{
 24ExampleHelper.SetConsoleLogger();
 25
 26// 申明一個主迴路排程組
 27var dispatcher = new DispatcherEventLoopGroup();
 28 
 29/*
 30Netty 提供了許多不同的 EventLoopGroup 的實現用來處理不同的傳輸。
 31在這個例子中我們實現了一個服務端的應用，因此會有2個 NioEventLoopGroup 會被使用。
 32第一個經常被叫做‘boss’，用來接收進來的連線。第二個經常被叫做‘worker’，用來處理已經被接收的連線，一旦‘boss’接收到連線，就會把連線資訊註冊到‘worker’上。
 33如何知道多少個執行緒已經被使用，如何對映到已經建立的 Channel上都需要依賴於 IEventLoopGroup 的實現，並且可以通過建構函式來配置他們的關係。
 34*/
 35 
 36// 主工作執行緒組，設定為1個執行緒
 37IEventLoopGroup bossGroup = dispatcher; // (1)
 38// 子工作執行緒組，設定為1個執行緒
 39IEventLoopGroup workerGroup = new WorkerEventLoopGroup(dispatcher);
 40 
 41try
 42{
 43// 宣告一個服務端Bootstrap，每個Netty服務端程式，都由ServerBootstrap控制，通過鏈式的方式組裝需要的引數
 44var serverBootstrap = new ServerBootstrap(); // (2)
 45// 設定主和工作執行緒組
 46serverBootstrap.Group(bossGroup, workerGroup);
 47 
 48if (ServerSettings.UseLibuv)
 49{
 50// 申明服務端通訊通道為TcpServerChannel
 51serverBootstrap.Channel<TcpServerChannel>(); // (3)
 52}
 53 
 54serverBootstrap
 55// 設定網路IO引數等
 56.Option(ChannelOption.SoBacklog, 100) // (5)
 57 
 58// 在主執行緒組上設定一個列印日誌的處理器
 59.Handler(new LoggingHandler("SRV-LSTN"))
 60 
 61// 設定工作執行緒引數
 62.ChildHandler(
 63/*
 64* ChannelInitializer 是一個特殊的處理類，他的目的是幫助使用者配置一個新的 Channel。
 65* 也許你想通過增加一些處理類比如DiscardServerHandler 來配置一個新的 Channel 或者其對應的ChannelPipeline 來實現你的網路程式。
 66* 當你的程式變的複雜時，可能你會增加更多的處理類到 pipline 上，然後提取這些匿名類到最頂層的類上。
 67*/
 68new ActionChannelInitializer<IChannel>( // (4)
 69channel =>
 70{
 71/*
 72* 工作執行緒聯結器是設定了一個管道，服務端主執行緒所有接收到的資訊都會通過這個管道一層層往下傳輸，
 73* 同時所有出棧的訊息 也要這個管道的所有處理器進行一步步處理。
 74*/
 75IChannelPipeline pipeline = channel.Pipeline;
 76 
 77// 新增日誌攔截器
 78pipeline.AddLast(new LoggingHandler("SRV-CONN"));
 79 
 80// 添加出棧訊息，通過這個handler在訊息頂部加上訊息的長度。
 81// LengthFieldPrepender(2)：使用2個位元組來儲存資料的長度。
 82pipeline.AddLast("framing-enc", new LengthFieldPrepender(2));
 83 
 84/*
 85入棧訊息通過該Handler,解析訊息的包長資訊，並將正確的訊息體傳送給下一個處理Handler
 861，InitialBytesToStrip = 0,//讀取時需要跳過的位元組數
 872，LengthAdjustment = -5,//包實際長度的糾正，如果包長包括包頭和包體，則要減去Length之前的部分
 883，LengthFieldLength = 4,//長度欄位的位元組數 整型為4個位元組
 894，LengthFieldOffset = 1,//長度屬性的起始（偏移）位
 905，MaxFrameLength = int.MaxValue, //最大包長
 91*/
 92pipeline.AddLast("framing-dec", new LengthFieldBasedFrameDecoder(ushort.MaxValue, 0, 2, 0, 2));
 93 
 94// 業務handler
 95pipeline.AddLast("echo", new EchoServerHandler());
 96}));
 97 
 98// bootstrap繫結到指定埠的行為就是服務端啟動服務，同樣的Serverbootstrap可以bind到多個埠
 99IChannel boundChannel = await serverBootstrap.BindAsync(ServerSettings.Port); // (6)
100 
101Console.WriteLine("wait the client input");
102Console.ReadLine();
103 
104// 關閉服務
105await boundChannel.CloseAsync();
106}
107finally
108{
109// 釋放指定工作組執行緒
110await Task.WhenAll( // (7)
111bossGroup.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1)),
112workerGroup.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1))
113);
114}
115}
116 
117static void Main() => RunServerAsync().Wait();
118}
119 }

View Code

1. IEventLoopGroup 是用來處理I/O操作的多執行緒事件迴圈器，DotNetty 提供了許多不同的 EventLoopGroup 的實現用來處理不同的傳輸。在這個例子中我們實現了一個服務端的應用，因此會有2個 IEventLoopGroup 會被使用。第一個經常被叫做‘boss’，用來接收進來的連線。第二個經常被叫做‘worker’，用來處理已經被接收的連線，一旦‘boss’接收到連線，就會把連線資訊註冊到‘worker’上。

2. ServerBootstrap 是一個啟動 Transport 服務的輔助啟動類。你可以在這個服務中直接使用 Channel，但是這會是一個複雜的處理過程，在很多情況下你並不需要這樣做。

3. 這裡我們指定使用 TcpServerChannel類來舉例說明一個新的 Channel 如何接收進來的連線。

4. ChannelInitializer 是一個特殊的處理類，他的目的是幫助使用者配置一個新的 Channel,當你的程式變的複雜時，可能你會增加更多的處理類到 pipline 上，然後提取這些匿名類到最頂層的類上。

5. 你可以設定這裡指定的 Channel 實現的配置引數。我們正在寫一個TCP/IP 的服務端，因此我們被允許設定 socket 的引數選項比如tcpNoDelay 和 keepAlive。

6. 繫結埠然後啟動服務，這裡我們在機器上綁定了機器網絡卡上的設定埠，當然現在你可以多次呼叫 bind() 方法(基於不同繫結地址)。

7. 使用完成後，優雅的釋放掉指定的工作組執行緒，當然，你可以選擇關閉程式，但這並不推薦。

Server端的事件處理程式碼：

上一部分程式碼中加粗地方的實現

 1 namespace Echo.Server
 2 {
 3using System;
 4using System.Text;
 5using DotNetty.Buffers;
 6using DotNetty.Transport.Channels;
 7 
 8/// <summary>
 9/// 服務端處理事件函式
10/// </summary>
11public class EchoServerHandler : ChannelHandlerAdapter // ChannelHandlerAdapter 業務繼承基類介面卡 // (1)
12{
13/// <summary>
14/// 管道開始讀
15/// </summary>
16/// <param name="context"></param>
17/// <param name="message"></param>
18public override void ChannelRead(IChannelHandlerContext context, object message) // (2)
19{
20if (message is IByteBuffer buffer)// (3)
21{
22Console.WriteLine("Received from client: " + buffer.ToString(Encoding.UTF8));
23}
24 
25context.WriteAsync(message); // (4)
26}
27 
28/// <summary>
29/// 管道讀取完成
30/// </summary>
31/// <param name="context"></param>
32public override void ChannelReadComplete(IChannelHandlerContext context) => context.Flush(); // (5)
33 
34/// <summary>
35/// 出現異常
36/// </summary>
37/// <param name="context"></param>
38/// <param name="exception"></param>
39public override void ExceptionCaught(IChannelHandlerContext context, Exception exception)
40{
41Console.WriteLine("Exception: " + exception);
42context.CloseAsync();
43}
44}
45 }

View Code

1. DiscardServerHandler 繼承自 ChannelInboundHandlerAdapter，這個類實現了IChannelHandler介面，IChannelHandler提供了許多事件處理的介面方法，然後你可以覆蓋這些方法。現在僅僅只需要繼承 ChannelInboundHandlerAdapter 類而不是你自己去實現介面方法。

2. 這裡我們覆蓋了 chanelRead() 事件處理方法。每當從客戶端收到新的資料時，這個方法會在收到訊息時被呼叫，這個例子中，收到的訊息的型別是 ByteBuf。

3. 為了響應或顯示客戶端發來的資訊，為此，我們將在控制檯中打印出客戶端傳來的資料。

4. 然後，我們將客戶端傳來的訊息通過context.WriteAsync寫回到客戶端。

5. 當然，步驟4只是將流快取到上下文中，並沒執行真正的寫入操作，通過執行Flush將流資料寫入管道，並通過context傳回給傳來的客戶端。

Client端程式碼：

重點看註釋的地方，其他地方跟Server端沒有任何區別

 1 namespace Echo.Client
 2 {
 3using System;
 4using System.Net;
 5using System.Text;
 6using System.Threading.Tasks;
 7using DotNetty.Buffers;
 8using DotNetty.Codecs;
 9using DotNetty.Handlers.Logging;
10using DotNetty.Transport.Bootstrapping;
11using DotNetty.Transport.Channels;
12using DotNetty.Transport.Channels.Sockets;
13using Examples.Common;
14 
15static class Program
16{
17static async Task RunClientAsync()
18{
19ExampleHelper.SetConsoleLogger();
20
21var group = new MultithreadEventLoopGroup();
22 
23try
24{
25var bootstrap = new Bootstrap();
26bootstrap
27.Group(group)
28.Channel<TcpSocketChannel>()
29.Option(ChannelOption.TcpNodelay, true)
30.Handler(
31new ActionChannelInitializer<ISocketChannel>(
32channel =>
33{
34IChannelPipeline pipeline = channel.Pipeline;
35pipeline.AddLast(new LoggingHandler());
36pipeline.AddLast("framing-enc", new LengthFieldPrepender(2));
37pipeline.AddLast("framing-dec", new LengthFieldBasedFrameDecoder(ushort.MaxValue, 0, 2, 0, 2));
38 
39pipeline.AddLast("echo", new EchoClientHandler());
40}));
41 
42IChannel clientChannel = await bootstrap.ConnectAsync(new IPEndPoint(ClientSettings.Host, ClientSettings.Port));
43
44// 建立死迴圈，類同於While(true)
45for (;;) // (4)
46{
47Console.WriteLine("input you data:");
48// 根據設定建立快取區大小
49IByteBuffer initialMessage = Unpooled.Buffer(ClientSettings.Size); // （1）
50string r = Console.ReadLine();
51// 將資料流寫入緩衝區
52initialMessage.WriteBytes(Encoding.UTF8.GetBytes(r ?? throw new InvalidOperationException())); // (2)
53// 將緩衝區資料流寫入到管道中
54await clientChannel.WriteAndFlushAsync(initialMessage); // (3)
55if(r.Contains("bye"))
56break;
57}
58 
59Console.WriteLine("byebye");
60
61 
62await clientChannel.CloseAsync();
63}
64finally
65{
66await group.ShutdownGracefullyAsync(TimeSpan.FromMilliseconds(100), TimeSpan.FromSeconds(1));
67}
68}
69 
70static void Main() => RunClientAsync().Wait();
71}
72 }

View Code

1. 初始化一個緩衝區的大小。

2. 預設緩衝區接受的資料型別為bytes[]，當然這樣也更加便於序列化成流。

3. 將緩衝區的流直接資料寫入到Channel管道中。該管道一般為連結通訊的另一端（C端）。

4. 建立死迴圈，這樣做的目的是為了測試每次都必須從客戶端輸入的資料，通過服務端迴路一次後，再進行下一次的輸入操作。

Client端的事件處理程式碼：

 1 namespace Echo.Client
 2 {
 3using System;
 4using System.Text;
 5using DotNetty.Buffers;
 6using DotNetty.Transport.Channels;
 7 
 8public class EchoClientHandler : ChannelHandlerAdapter
 9{
10readonly IByteBuffer initialMessage;
11 
12public override void ChannelActive(IChannelHandlerContext context) => context.WriteAndFlushAsync(this.initialMessage);
13 
14public override void ChannelRead(IChannelHandlerContext context, object message)
15{
16if (message is IByteBuffer byteBuffer)
17{
18Console.WriteLine("Received from server: " + byteBuffer.ToString(Encoding.UTF8));
19}
20}
21 
22public override void ChannelReadComplete(IChannelHandlerContext context) => context.Flush();
23 
24public override void ExceptionCaught(IChannelHandlerContext context, Exception exception)
25{
26Console.WriteLine("Exception: " + exception);
27context.CloseAsync();
28}
29}
30 }

View Code

非常簡單，將資料流顯示到控制檯。

實現結果

至此，我們使用DotNetty框架搭建簡單的應答伺服器就這樣做好了，很簡單，實現效果如下：

C端主動向S端主動傳送資料後，S端收到資料，在控制檯打印出資料，並回傳給C端，當然，S端還可以做很多很多的事情。

DotNetty內部除錯記錄分析

雖然DotNetty官方沒有提供任何技術文件，但官方卻提供了詳細的除錯記錄，很多時候，我們學習者其實也可以通過除錯記錄來分析某一個功能的實現流程。我們可以通過將DotNetty的內部輸入輸出記錄列印到控制檯上。

InternalLoggerFactory.DefaultFactory.AddProvider(new ConsoleLoggerProvider((s, level) => true, false));

可以看到服務端的列印記錄一下多出來了許多許多，有大部分是屬於DotNetty內部除錯時的列印記錄，我們只著重看如下的部分。

dbug: SRV-LSTN[0]
[id: 0x3e8afca1] HANDLER_ADDED
dbug: SRV-LSTN[0]
[id: 0x3e8afca1] REGISTERED （1）
dbug: SRV-LSTN[0]
[id: 0x3e8afca1] BIND: 0.0.0.0:8007 （2）
wait the client input
dbug: SRV-LSTN[0]
[id: 0x3e8afca1, 0.0.0.0:8007] ACTIVE （3）
dbug: SRV-LSTN[0]
[id: 0x3e8afca1, 0.0.0.0:8007] READ （4）
dbug: SRV-LSTN[0]
[id: 0x3e8afca1, 0.0.0.0:8007] RECEIVED: [id: 0x7bac2775, 127.0.0.1:64073 :> 127.0.0.1:8007] （5）
dbug: SRV-LSTN[0]
[id: 0x3e8afca1, 0.0.0.0:8007] RECEIVED_COMPLETE （6）
dbug: SRV-LSTN[0]
[id: 0x3e8afca1, 0.0.0.0:8007] READ （7）
dbug: SRV-CONN[0]
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] HANDLER_ADDED （8）
dbug: SRV-CONN[0]
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] REGISTERED （9）
dbug: SRV-CONN[0]
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] ACTIVE （10）
dbug: SRV-CONN[0]
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] READ （11）
dbug: DotNetty.Buffers.AbstractByteBuffer[0]（12）
-Dio.netty.buffer.bytebuf.checkAccessible: True
dbug: SRV-CONN[0]
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] RECEIVED: 14B （13）
+-------------------------------------------------+
|0123456789abcdef |
+--------+-------------------------------------------------+----------------+
|100000000| 00 0C 68 65 6C 6C 6F 20 77 6F 72 6C 64 21|..hello world!|
+--------+-------------------------------------------------+----------------+
Received from client: hello world!
dbug: SRV-CONN[0]（14）
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] WRITE: 2B
+-------------------------------------------------+
|0123456789abcdef |
+--------+-------------------------------------------------+----------------+
|100000000| 00 0C|..|
+--------+-------------------------------------------------+----------------+
dbug: SRV-CONN[0] （15）
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] WRITE: 12B
+-------------------------------------------------+
|0123456789abcdef |
+--------+-------------------------------------------------+----------------+
|100000000| 68 65 6C 6C 6F 20 77 6F 72 6C 64 21|hello world!|
+--------+-------------------------------------------------+----------------+
dbug: SRV-CONN[0] （16）
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] RECEIVED_COMPLETE
dbug: SRV-CONN[0] （17）
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] FLUSH
dbug: SRV-CONN[0] （18）
[id: 0x7bac2775, 127.0.0.1:64073 => 127.0.0.1:8007] READ

咋一看，有18個操作，好像有點太多了，其實不然，還有很多很多的內部除錯細節並沒列印到控制檯上。

1. 通過手動建立的工作執行緒組，並將這組執行緒註冊到管道中，這個管道可以是基於SOCKER，可以基於IChannel（1）；

2. 繫結自定的IP地址和埠號到自定義管道上（2）；

3. 啟用自定義管道（3）；

4. 開始讀取（其實也是開始監聽）（4）；

5. 收到來自id為0x7bac2775的客戶端連線請求，建立連線，並繼續開始監聽（5）（6）（7）；

6. 從第8步開始，日誌已經變成id為0x7bac2775的記錄了，當然一樣包含註冊管道，啟用管道，開始監聽等等與S端一模一樣的操作（8）（9）（10）（11）

7. 當筆者輸入一條"hello world!"資料後，DotNetty.Buffers.AbstractByteBuffer會進行資料型別檢查，以便確認能將資料放入到管道中。（12）

8. 將資料傳送到S端，資料大小為14B，hello world前有兩個點，代表這是資料頭，緊接著再發送兩個點，但沒有任何資料，代表資料已經結束。DotNetty將資料的十六進位制儲存位用易懂的方式表現了出來，很人性化。（13）（14）

9. S端收到資料沒有任何加工和處理，馬上將資料回傳到C端。（15）（16）

10. 最後，當這個過程完成後，需要將快取區的資料強制寫入到管道中，所以會執行一次Flush操作，整個傳輸完成。接下來，不管是C端還是S端，繼續將自己的狀態改成READ，用於監聽管道中的各種情況，比如連線狀態，資料傳輸等等（17）。

總結

對於剛開始接觸Socket程式設計的朋友而言，這是個噩夢，因為Socket程式設計的複雜性不會比多執行緒容易，甚至會更復雜。協議，壓縮，傳輸，多執行緒，監聽，流控制等等一系列問題擺在面前，因此而誕生了Netty這樣優秀的開源框架，但是Netty是個半成品，因為你需要基於他來實現自己想要的協議，傳輸等等自定義操作，而底層的內容，你完全不用關心。不像某些框架，比如Newtonsoft.Json這樣的功能性框架，不用配置，不用自定義，直接拿來用就可以了。

雖然DotNetty幫我們實現了底層大量的操作，但如果不熟悉或者一點也不懂網路通訊，同樣對上面的程式碼是一頭霧水，為何？行情需要，我們程式設計師天天都在趕業務，哪有時間去了解和學習更多的細節...通過將除錯記錄打印出來，並逐行挨個的對照程式碼進行分析，就會慢慢開始理解最簡單的通訊流程了。

本篇只是實現了基於DotNetty最簡單的通訊過程，也只是將資料做了一下回路，並沒做到任何與RPC有關的呼叫，下一篇我們開始講這個例子深入，介紹基於DotNetty的RPC呼叫。

碼字不易，感謝閱讀！