哎~~ 想想大部分園友應該對 "高性能" 字樣更感興趣，為了吸引眼球所以標題中一定要突出，其實我更喜歡的標題是《猴賽雷，C#編寫TCP服務的花樣姿勢！》。
本篇文章的主旨是使用 .NET/C# 實現 TCP 高性能服務的不同方式，包括但不限於如下內容：
在 .NET/C# 中對於 Socket 的支持均是基於 Windows I/O Completion Ports 完成端口技術的封裝，通過不同的 Non-Blocking 封裝結構來滿足不同的編程需求。以上方式均已在 Cowboy.Sockets 中有完整實現，並且 APM 和 TAP 方式已經在實際項目中應用。Cowboy.Sockets 還在不斷的進化和完善中，如有任何問題請及時指正。
雖然有這麼多種實現方式，但抽象的看，它們是一樣一樣的，用兩個 Loop 即可描述：Accept Loop 和 Read Loop，如下圖所示。（這裡提及的 "Loop" 指的是一種循環方式，而非特指 while/for 等關鍵字。）
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));
}
AsyncTcpSocketServer 的實現是基於 .NET Framework 自帶的 TcpListener 和 TcpClient 的更進一步的封裝，采用基於 TAP 的 async/await 的 XXXAsync 接口實現。
然而，實際上 XXXAsync 並沒有創建什麼神奇的效果，其內部實現只是將 APM 的方法轉換成了 TAP 的調用方式。
//************* 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);
}
AsyncTcpSocketServer 中的 Accept Loop 指的就是，
while (IsListening)
{
var tcpClient = await _listener.AcceptTcpClientAsync();
}
每一個建立成功的 Connection 由 AsyncTcpSocketSession 來處理，所以 AsyncTcpSocketSession 中會包含 Read Loop，
while (State == TcpSocketConnectionState.Connected)
{
int receiveCount = await _stream.ReadAsync(_receiveBuffer, 0, _receiveBuffer.Length);
}
為了將 async/await 異步到底，AsyncTcpSocketServer 所暴露的接口也同樣是 Awaitable 的。
public interface IAsyncTcpSocketServerMessageDispatcher
{
Task OnSessionStarted(AsyncTcpSocketSession session);
Task OnSessionDataReceived(AsyncTcpSocketSession session, byte[] data, int offset, int count);
Task OnSessionClosed(AsyncTcpSocketSession session);
}
使用時僅需將一個實現了該接口的對象注入到 AsyncTcpSocketServer 的構造函數中即可。
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 是 SocketAsyncEventArgs 的簡寫。SocketAsyncEventArgs 是 .NET Framework 3.5 開始支持的一種支持高性能 Socket 通信的實現。SocketAsyncEventArgs 相比於 APM 方式的主要優點可以描述如下：
The main feature of these enhancements is the avoidance of the repeated allocation and synchronization of objects during high-volume asynchronous socket I/O. The Begin/End design pattern currently implemented by the Socket class for asynchronous socket I/O requires a System.IAsyncResult object be allocated for each asynchronous socket operation.
也就是說，優點就是無需為每次調用都生成 IAsyncResult 等對象，向原生 Socket 更靠近一些。
使用 SocketAsyncEventArgs 的推薦步驟如下：
重點在於池化（Pooling），池化的目的就是為了重用和減少運行時分配和垃圾回收的壓力。
TcpSocketSaeaServer 即是對 SocketAsyncEventArgs 的應用和封裝，並實現了 Pooling 技術。TcpSocketSaeaServer 中的重點是 SaeaAwaitable 類，SaeaAwaitable 中內置了一個 SocketAsyncEventArgs，並通過 GetAwaiter 返回 SaeaAwaiter 來支持 async/await 操作。同時，通過 SaeaExtensions 擴展方法對來擴展 SocketAsyncEventArgs 的 Awaitable 實現。
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<SaeaAwaitable> 的衍生實現，用於池化 SaeaAwaitable 實例。同時，為了減少 TcpSocketSaeaSession 的構建過程，也實現了 SessionPool 即 QueuedObjectPool<TcpSocketSaeaSession>。
TcpSocketSaeaServer 中的 Accept Loop 指的就是，
while (IsListening)
{
var saea = _acceptSaeaPool.Take();
var socketError = await _listener.AcceptAsync(saea);
if (socketError == SocketError.Success)
{
var acceptedSocket = saea.Saea.AcceptSocket;
}
_acceptSaeaPool.Return(saea);
}
每一個建立成功的 Connection 由 TcpSocketSaeaSession 來處理，所以 TcpSocketSaeaSession 中會包含 Read Loop，
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;
}
同樣，TcpSocketSaeaServer 對外所暴露的接口也同樣是 Awaitable 的。
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;
}
}
從 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.
到目前為止，.NET Framework 還沒有推出對 RIO 的支持，所以若想在 C# 中實現 RIO 則只能通過 P/Invoke 方式，RioSharp 是開源項目中的一個比較完整的實現。
Cowboy.Sockets 直接引用了 RioSharp 的源代碼，放置在 Cowboy.Sockets.Experimental 名空間下，以供實驗和測試使用。
同樣，通過 TcpSocketRioServer 來實現 Accept Loop，
_listener.OnAccepted = (acceptedSocket) =>
{
Task.Run(async () =>
{
await Process(acceptedSocket);
})
.Forget();
};
通過 TcpSocketRioSession 來處理 Read Loop，
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服務的多種實現方式》由 Dennis Gao 發表自博客園個人博客，未經作者本人同意禁止以任何的形式轉載，任何自動的或人為的爬蟲轉載行為均為耍流氓。
