這篇文章的主題是使用 .NET/C# 實現TCP 高效能服務的不同方式,包含但不限於以下內容:
TAP 方式,即Task-based Asynchronous Pattern
#SAEA 方式,即SocketAsyncEventArgs
RIO 方式,即Registered I/O
#在.NET/C# 中對於Socket 的支援皆是基於Windows I/O Completion Ports 完成連接埠技術的封裝,透過不同的Non-Blocking 封裝結構來滿足不同的程式設計需求。以上方式均已在 Cowboy.Sockets 中有完整實現,並且 APM 和 TAP 方式已經在實際專案中應用。 Cowboy.Sockets 仍在不斷的進化和完善中,如有任何問題請及時指正。
雖然有這麼多實作方式,但抽象的看,它們是一樣一樣的,用兩個Loop 即可描述:Accept Loop 和Read Loop ,如下圖所示。 (這裡提及的"Loop" 指的是一種循環方式,而非特指 while/for 等關鍵字。以建立TCP Connection。
在任何 TCP Server 的實作中,一定存在一個 Read Socket Loop,用來接收 Client 端 Write 過來的資料。Time
out 的異常。如果 Read 迴圈阻塞,顯然會導致無法及時收到 Client 端發過來的數據,進而導致 Client 端 Send Buffer 滿,無法再傳送資料。#Accept 到新的 Socket,建構新的Connection 需要分配各種資源,分配資源慢;
Accept 到新的 Socket,沒有及時觸發下一次Accept;Java 中著名的Netty 網路庫從4.0 版本開始對Buffer 部分做了全新的嘗試,採用了名叫ByteBuf 的設計,實現Buffer Zero Copy 以減少高並發條件下 Buffer 拷貝帶來的效能損失和GC 壓力。 DotNetty,Orleans ,Helios 等專案正在嘗試在 C# 中進行類似的 ByteBuf 的實作。 APM 方法:TcpSocketServer
##li早期字##實現。 TcpSocketServer 中的Accept Loop 指的就是,
來處理,所以TcpSocketSession 會包含Read Loop, BeginRead -> EndRead -> BeginRead -> EndRead -> BeginRead -> ... TcpSocketServer 透過揭露Event 來實現Connection 的建立與斷開和資料接收的通知。 使用也是簡單直接,直接訂閱事件通知。 AsyncTcpSocketServer 的實作是基於.NET Framework 內建的 TcpListener 和 TcpClient 的更進一步的封裝,採用基於TAP 的async/await 的XXXAsync 介面實作。 然而,實際上 XXXAsync 並沒有創造出什麼神奇的效果,其內部實作只是將 APM 的方法轉換成了 TAP 的呼叫方式。 AsyncTcpSocketServer 中的Accept Loop 指的就是, 每一個建立成功的Connection由AsyncTcpSocketSession# 來處理,所以AsyncTcpSocketSession 中會包含Read Loop, 為了將 async/await 非同步到底,AsyncTcpSocketServer 所暴露的介面也同樣是Awaitable 的。 使用時只需將一個實作了該介面的物件注入到 AsyncTcpSocketServer 的建構子中即可。 當然,對於介面的實作也不是強制了,也可以在建構函式中直接注入方法的實作。 SAEA 是 SocketAsyncEventArgs 的簡寫。 SocketAsyncEventArgs 是 .NET Framework 3.5 開始支援的支援高效能 Socket 通訊的實作。 SocketAsyncEventArgs 相較於APM 方式的主要優點可以描述如下: The main feature of these enhancements is the avoidance of the repeated all ocation and synchronization of objects during high-volume asynchronous socket I/O. The Begin/End design pattern currently implemented by the Socket class for asynchronous socketly implemented by the Socket class for asynchronous socket I/## s a System.IAsyncResult object be allocated for each asynchronous socket operation. 重點在於池化(Pooling),池化的目的就是為了重複使用和減少運行時分配和垃圾回收的壓力。 TcpSocketSaeaServer 即是 SocketAsyncEventArgs 的應用與封裝,並實作了 Pooling 技術。 TcpSocketSaeaServer 中的重點是 SaeaAwaitable 類,SaeaAwaitable 中內建了一個 SocketAsyncEventArgs,並透過 GetAwaiter 返回 SaeaAwaiter 來支援 async/await 操作。同時,透過 SaeaExtensions 擴展方法對來擴展 SocketAsyncEventArgs 的 Awaitable 實作。 TcpSocketSaeaServer 中的 Accept Loop 指的就是, 每一个建立成功的 Connection 由 TcpSocketSaeaSession 来处理,所以 TcpSocketSaeaSession 中会包含 Read Loop, 同样,TcpSocketSaeaServer 对外所暴露的接口也同样是 Awaitable 的。 使用起来也是简单直接: 从 Windows 8.1 / Windows Server 2012 R2 开始,微软推出了 Registered I/O Networking Extensions 来支持高性能 Socket 服务的实现,简称 RIO。 The following functions are supported for Windows Store apps on Windows 8.1, Windows Server 2012 R2, and later. Microsoft Visual Studio 2013 Update 3 or later is required for Windows Store apps. RIOCloseCompletionQueue RIOCreateCompletionQueue RIOCreateRequestQueue RIODequeueCompletion RIODeregisterBuffer RIONotify RIOReceive RIOReceiveEx RIORegisterBuffer RIOResizeCompletionQueue RIOResizeRequestQueue RIOSend RIOSendEx 到目前为止,.NET Framework 还没有推出对 RIO 的支持,所以若想在 C# 中实现 RIO 则只能通过 P/Invoke 方式,RioSharp 是开源项目中的一个比较完整的实现。 Cowboy.Sockets 直接引用了 RioSharp 的源代码,放置在 Cowboy.Sockets.Experimental 名空间下,以供实验和测试使用。 同样,通过 TcpSocketRioServer 来实现 Accept Loop, 通过 TcpSocketRioSession 来处理 Read Loop, 测试代码一如既往的类似: Asynchronous Programming Model (APM) Task-based Asynchronous Pattern (TAP) Event-based Asynchronous Pattern (EAP) SocketAsyncEventArgs Registered I/O Netty: Reference counted objects Socket Performance Enhancements in Version 3.5 What's New for Windows Sockets for Windows 8.1 and Windows Server 2012 R2 RIO_EXTENSION_FUNCTION_TABLE structure Windows 8 Registered I/O Networking Extensions 本篇文章《C#高性能TCP服务的多种实现方式》由 Dennis Gao 发表自博客园个人博客,未经作者本人同意禁止以任何的形式转载,任何自动的或人为的爬虫转载行为均为耍流氓。 event EventHandler<TcpClientConnectedEventArgs> ClientConnected; event EventHandler<TcpClientDisconnectedEventArgs> ClientDisconnected; event EventHandler<TcpClientDataReceivedEventArgs> ClientDataReceived;
private static void StartServer()
{
_server = new TcpSocketServer(22222);
_server.ClientConnected += server_ClientConnected;
_server.ClientDisconnected += server_ClientDisconnected;
_server.ClientDataReceived += server_ClientDataReceived;
_server.Listen();
}
static void server_ClientConnected(object sender, TcpClientConnectedEventArgs e)
{
Console.WriteLine(string.Format("TCP client {0} has connected {1}.", e.Session.RemoteEndPoint, e.Session));
}
static void server_ClientDisconnected(object sender, TcpClientDisconnectedEventArgs e)
{
Console.WriteLine(string.Format("TCP client {0} has disconnected.", e.Session));
}
static void server_ClientDataReceived(object sender, TcpClientDataReceivedEventArgs e)
{ var text = Encoding.UTF8.GetString(e.Data, e.DataOffset, e.DataLength);
Console.Write(string.Format("Client : {0} {1} --> ", e.Session.RemoteEndPoint, e.Session));
Console.WriteLine(string.Format("{0}", text));
_server.Broadcast(Encoding.UTF8.GetBytes(text));
}
TAP 方法:AsyncTcpSocketServer
//************* Task-based async public methods *************************
[HostProtection(ExternalThreading = true)] public Task<Socket> AcceptSocketAsync()
{ return Task<Socket>.Factory.FromAsync(BeginAcceptSocket, EndAcceptSocket, null);
}
[HostProtection(ExternalThreading = true)] public Task<TcpClient> AcceptTcpClientAsync()
{ return Task<TcpClient>.Factory.FromAsync(BeginAcceptTcpClient, EndAcceptTcpClient, null);
}
while (IsListening)
{ var tcpClient = await _listener.AcceptTcpClientAsync();
}
while (State == TcpSocketConnectionState.Connected)
{ int receiveCount = await _stream.ReadAsync(_receiveBuffer, 0, _receiveBuffer.Length);
}
public interface IAsyncTcpSocketServerMessageDispatcher
{
Task OnSessionStarted(AsyncTcpSocketSession session);
Task OnSessionDataReceived(AsyncTcpSocketSession session, byte[] data, int offset, int count);
Task OnSessionClosed(AsyncTcpSocketSession session);
}
public class SimpleMessageDispatcher : IAsyncTcpSocketServerMessageDispatcher
{ public async Task OnSessionStarted(AsyncTcpSocketSession session)
{
Console.WriteLine(string.Format("TCP session {0} has connected {1}.", session.RemoteEndPoint, session)); await Task.CompletedTask;
}
public async Task OnSessionDataReceived(AsyncTcpSocketSession session, byte[] data, int offset, int count)
{ var text = Encoding.UTF8.GetString(data, offset, count);
Console.Write(string.Format("Client : {0} --> ", session.RemoteEndPoint));
Console.WriteLine(string.Format("{0}", text));
await session.SendAsync(Encoding.UTF8.GetBytes(text));
}
public async Task OnSessionClosed(AsyncTcpSocketSession session)
{
Console.WriteLine(string.Format("TCP session {0} has disconnected.", session)); await Task.CompletedTask;
}
}
public AsyncTcpSocketServer(
IPEndPoint listenedEndPoint,
Func<AsyncTcpSocketSession, byte[], int, int, Task> onSessionDataReceived = null,
Func<AsyncTcpSocketSession, Task> onSessionStarted = null,
Func<AsyncTcpSocketSession, Task> onSessionClosed = null,
AsyncTcpSocketServerConfiguration configuration = null)
{}
SAEA 方式:TcpSocketSaeaServer
也就是說,優點就是無需為每次調用生成 IAsyncResult 等對象,向原生Socket 更靠近一些。 使用SocketAsyncEventArgs 的推薦步驟如下:
public static SaeaAwaitable AcceptAsync(this Socket socket, SaeaAwaitable awaitable) public static SaeaAwaitable ConnectAsync(this Socket socket, SaeaAwaitable awaitable) public static SaeaAwaitable DisonnectAsync(this Socket socket, SaeaAwaitable awaitable) public static SaeaAwaitable ReceiveAsync(this Socket socket, SaeaAwaitable awaitable) public static SaeaAwaitable SendAsync(this Socket socket, SaeaAwaitable awaitable)
###SaeaPool 則是一個 QueuedObjectPool while (IsListening)
{ var saea = _acceptSaeaPool.Take();
var socketError = await _listener.AcceptAsync(saea); if (socketError == SocketError.Success)
{ var acceptedSocket = saea.Saea.AcceptSocket;
}
_acceptSaeaPool.Return(saea);
}
var saea = _saeaPool.Take();
saea.Saea.SetBuffer(_receiveBuffer, 0, _receiveBuffer.Length);
while (State == TcpSocketConnectionState.Connected)
{
saea.Saea.SetBuffer(0, _receiveBuffer.Length);
var socketError = await _socket.ReceiveAsync(saea); if (socketError != SocketError.Success) break;
var receiveCount = saea.Saea.BytesTransferred; if (receiveCount == 0) break;
}
public interface ITcpSocketSaeaServerMessageDispatcher
{
Task OnSessionStarted(TcpSocketSaeaSession session);
Task OnSessionDataReceived(TcpSocketSaeaSession session, byte[] data, int offset, int count);
Task OnSessionClosed(TcpSocketSaeaSession session);
}
public class SimpleMessageDispatcher : ITcpSocketSaeaServerMessageDispatcher
{ public async Task OnSessionStarted(TcpSocketSaeaSession session)
{
Console.WriteLine(string.Format("TCP session {0} has connected {1}.", session.RemoteEndPoint, session)); await Task.CompletedTask;
}
public async Task OnSessionDataReceived(TcpSocketSaeaSession session, byte[] data, int offset, int count)
{ var text = Encoding.UTF8.GetString(data, offset, count);
Console.Write(string.Format("Client : {0} --> ", session.RemoteEndPoint));
Console.WriteLine(string.Format("{0}", text));
await session.SendAsync(Encoding.UTF8.GetBytes(text));
}
public async Task OnSessionClosed(TcpSocketSaeaSession session)
{
Console.WriteLine(string.Format("TCP session {0} has disconnected.", session)); await Task.CompletedTask;
}
}
RIO 方式:TcpSocketRioServer
_listener.OnAccepted = (acceptedSocket) =>{
Task.Run(async () =>
{ await Process(acceptedSocket);
})
.Forget();
};
while (State == TcpSocketConnectionState.Connected)
{ int receiveCount = await _stream.ReadAsync(_receiveBuffer, 0, _receiveBuffer.Length); if (receiveCount == 0) break;
}
public class SimpleMessageDispatcher : ITcpSocketRioServerMessageDispatcher
{ public async Task OnSessionStarted(TcpSocketRioSession session)
{ //Console.WriteLine(string.Format("TCP session {0} has connected {1}.", session.RemoteEndPoint, session));
Console.WriteLine(string.Format("TCP session has connected {0}.", session)); await Task.CompletedTask;
}
public async Task OnSessionDataReceived(TcpSocketRioSession session, byte[] data, int offset, int count)
{ var text = Encoding.UTF8.GetString(data, offset, count); //Console.Write(string.Format("Client : {0} --> ", session.RemoteEndPoint));
Console.Write(string.Format("Client : --> "));
Console.WriteLine(string.Format("{0}", text));
await session.SendAsync(Encoding.UTF8.GetBytes(text));
}
public async Task OnSessionClosed(TcpSocketRioSession session)
{
Console.WriteLine(string.Format("TCP session {0} has disconnected.", session)); await Task.CompletedTask;
}
}
参考资料
以上是C# 高效能 TCP 服務的多種實現的詳細內容。更多資訊請關注PHP中文網其他相關文章!