一、前言
时间过得真是快，转眼就2018年了。首先祝各位博友，软件开发者新年新气象，事业有成，身体健康，阖家幸福！最近看到园子里好多关于自己的2017年度总结以及对自己新一年的愿景，觉得咱园子的氛围是真的好。这三天假期我也没闲着，一边看OB海鲜团吃鸡一边写Socket SocketAsyncEventArgs 代码。我上一篇博客已经用APM的方式实现了客户端与服务器端的Socket通信，并具有了一定的并发能力。所以这三天我就决定对服务器代码进行改造，使用MS在4.0时发布的SocketAsyncEventArgs（SAEA）写法。为了方便的进行服务器端两种写法的对比，我客户端的代码没有进行变化，依然使用APM方式。代码已经上传至Github，链接会在文末贴出。
二、我的业务功能
我的业务功能依然是实现从服务器多线程下载更新文件。下载之前的那些操作我基本就不讲了，上一篇博文里的都有，本文还是回到Socket下载文件上。具体流程如下：
在我写SAEA代码之前，我仔细搜了一下网上的资源：MSDN、CNBLOG、CSDN、CodeProject。这四种来源的代码示例的主要流程是这样的：
对比我的流程，您会发现少了一半的通信过程。客户端的代码好写，但是服务器端如何发送完数据之后再接收数据？这中间的衔接过程还是有点门道的。特别是SAEA的代码采用了Buffer池化以及SAEA池化之后，里面有些小的细节就要想清楚了。下面就是具体的代码，我会以我自己的视角去论述APM与SAEA到底有什么区别。
三、对比
其实对于服务器端的APM，我觉得最重要的并不是代码中的BeginXXX或者是EndXXX，因为这就是APM写法的特征，BeginXXX或者EndXXX然后里面有一个回调函数，在回调函数里去做一些业务上的事情。最重要的是要有一个线程等待的概念，也就是代码中的ManualResetEvent这个东西，它就像地铁闸机一样，处理好一个再放一个进去。APM写法的好处是显而易见的，就是代码看起来十分的简单。缺点依照MS的说法就是如果有过多的客-服交流，可能会产生较多的IAsyncResult对象，这样会增加服务器的开销。
服务器端的APM写法：
using System;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using UpdaterShare.GlobalSetting;
using UpdaterShare.Model;
using UpdaterShare.Utility; namespace UpdaterServerAPM
{
public static class ServerSocket
{
private static int _downloadChannelsCount;
private static string _serverPath;
private static readonly ManualResetEvent AllDone = new ManualResetEvent(false); public static void StartServer(int port, int backlog)
{
_downloadChannelsCount = DownloadSetting.DownloadChannelsCount;
try
{
IPAddress ipAddress = IPAddress.Any;
IPEndPoint localEndPoint = new IPEndPoint(ipAddress, port);
Socket listener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
listener.Bind(localEndPoint);
listener.Listen(backlog); while (true)
{
AllDone.Reset();
listener.BeginAccept(AcceptCallback, listener);
AllDone.WaitOne();
}
}
catch (Exception ex)
{
var path = $"{AppDomain.CurrentDomain.BaseDirectory}\\RunLog.txt";
File.AppendAllText(path, ex.Message);
}
} private static void AcceptCallback(IAsyncResult ar)
{
AllDone.Set();
Socket listener = (Socket)ar.AsyncState;
Socket handler = listener.EndAccept(ar);
ComObject state = new ComObject { WorkSocket = handler };
handler.BeginReceive(state.Buffer, , ComObject.BufferSize, , FindUpdateFileCallback, state);
} private static void FindUpdateFileCallback(IAsyncResult ar)
{
ComObject state = (ComObject)ar.AsyncState;
Socket handler = state.WorkSocket;
int bytesRead = handler.EndReceive(ar);
if (bytesRead > )
{
var receiveData = state.Buffer.Take(bytesRead).ToArray();
var dataList = PacketUtils.SplitBytes(receiveData, PacketUtils.ClientFindFileInfoTag());
if (dataList != null && dataList.Any())
{
var request = PacketUtils.GetData(PacketUtils.ClientFindFileInfoTag(), dataList.FirstOrDefault());
string str = System.Text.Encoding.UTF8.GetString(request);
var infos = str.Split('_');
var productName = infos[];
var revitVersion = infos[];
var currentVersion = infos[]; var mainFolder = AppDomain.CurrentDomain.BaseDirectory.Replace("bin", "TestFile");
var serverFileFolder = Path.Combine(mainFolder, "Server");
var serverFileFiles = new DirectoryInfo(serverFileFolder).GetFiles(); var updatefile = serverFileFiles.FirstOrDefault(x=>x.Name.Contains(productName) && x.Name.Contains(revitVersion) && x.Name.Contains(currentVersion));
if (updatefile != null)
{
if (string.IsNullOrEmpty(updatefile.FullName) || !File.Exists(updatefile.FullName)) return;
_serverPath = updatefile.FullName;
FoundUpdateFileResponse(handler);
}
}
}
} private static void FoundUpdateFileResponse(Socket handler)
{
byte[] foundUpdateFileData = PacketUtils.PacketData(PacketUtils.ServerFoundFileInfoTag(),null);
ComObject state = new ComObject { WorkSocket = handler };
handler.BeginSend(foundUpdateFileData, , foundUpdateFileData.Length, , HasFoundUpdateFileCallback, state);
} private static void HasFoundUpdateFileCallback(IAsyncResult ar)
{
ComObject state = (ComObject)ar.AsyncState;
Socket handler = state.WorkSocket;
handler.EndSend(ar);
handler.BeginReceive(state.Buffer, , ComObject.BufferSize, , ReadFilePositionRequestCallback, state);
} private static void ReadFilePositionRequestCallback(IAsyncResult ar)
{
ComObject state = (ComObject)ar.AsyncState;
Socket handler = state.WorkSocket;
int bytesRead = handler.EndReceive(ar);
if (bytesRead > )
{
var receiveData = state.Buffer.Take(bytesRead).ToArray();
var dataList = PacketUtils.SplitBytes(receiveData, PacketUtils.ClientRequestFileTag());
if (dataList != null)
{
foreach (var request in dataList)
{
if (PacketUtils.IsPacketComplete(request))
{
int startPosition = PacketUtils.GetRequestFileStartPosition(request);
SendFileResponse(handler, startPosition);
}
}
}
}
} private static void SendFileResponse(Socket handler, int startPosition)
{
var packetSize = PacketUtils.GetPacketSize(_serverPath, _downloadChannelsCount);
if (packetSize != )
{
byte[] filedata = FileUtils.GetFile(_serverPath, startPosition, packetSize);
byte[] packetNumber = BitConverter.GetBytes(startPosition/packetSize);
if (filedata != null)
{
byte[] segmentedFileResponseData = PacketUtils.PacketData(PacketUtils.ServerResponseFileTag(), filedata, packetNumber);
ComObject state = new ComObject {WorkSocket = handler};
handler.BeginSend(segmentedFileResponseData, , segmentedFileResponseData.Length, , SendFileResponseCallback, state);
}
}
else
{
handler.Shutdown(SocketShutdown.Both);
handler.Close();
}
} private static void SendFileResponseCallback(IAsyncResult ar)
{
try
{
ComObject state = (ComObject)ar.AsyncState;
Socket handler = state.WorkSocket;
handler.EndSend(ar);
handler.Shutdown(SocketShutdown.Both);
handler.Close();
}
catch (Exception e)
{ }
}
}
}
说到SAEA，我觉得初入的小伙伴一定要先看MSDN上的实例，特别是它的BufferManager以及SocketAsyncEventArgsPool是怎么写的，到底是干什么用的。这里我可以简单的说下：SocketAsyncEventArgsPool是用来存放SAEA对象的，其个数依赖于你服务器所能承担的队列长度，比如说我服务器能承担100个客户的等待，我就在服务器端生成100个SAEA对象放在池子里，当有客户来连接时，我从池子里取出一个来和他对接。客户走了，我再扔到池子里去。BufferManager则是对池子里的SAEA对象进行Buffer分配的，也相当于一个池子，这个池子的大小是队列长度*通信缓存长度*2，乘以2是因为读与写是分开的。通信缓存长度很好理解，客户端要传个2G的信息给服务器端不可能一下子接收2G，肯定是一口一口吃，那么这一口的大小就是通信缓存长度。那么分配给每个SAEA的缓存是多大呢？当然就是通信缓存长度的大小咯。注意！！注意！！注意！！既然是池化了，所有关于Buffer的操作都要围绕分配给SAEA的Buffer去操作！见148-149行。当服务器拿着分配到的Buffer去接收信息后，如果再要发送信息，所要做的第一件事就是先清空分配的Buffer再使用，BufferManager给你分配哪段你就用哪段，别使用错了。有几个参数需要注意：e.Offset（偏移），e.Count（大小），e.Buffer（缓存字节数组）， e.BytesTransferred（通信传输的字节长度）。如果服务器端要发送数据，一定要用Array.Copy将信息写入对应分配的Buffer中。
说完池化，接着就是写法上的小区别，我觉得区别并不大，无非就是委托换了个写法。当然还要判断下是否为异步操作，如果是否则需要进行同步操作，见82-85行代码。
服务器的SAEA写法：
using System;
using System.IO;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using UpdaterShare.GlobalSetting;
using UpdaterShare.Model;
using UpdaterShare.Utility; namespace UpdaterServerSAEA
{
public class ServerSocket
{
private readonly int _port;
private readonly int _backlog;
private Socket _listenSocket;
private const int _opsToPreAlloc = ;
private readonly BufferManager _bufferManager;
private readonly SocketAsyncEventArgsPool _readWritePool;
private readonly Semaphore _maxNumberAcceptedClients; private string _serverPath;
private static readonly int _downloadChannelsCount = DownloadSetting.DownloadChannelsCount; public ServerSocket(int port, int backlog)
{
_port = port;
_backlog = backlog; _bufferManager = new BufferManager(ComObject.BufferSize * backlog * _opsToPreAlloc, ComObject.BufferSize);
_readWritePool = new SocketAsyncEventArgsPool(backlog);
_maxNumberAcceptedClients = new Semaphore(backlog, backlog);
} private void Init()
{
_bufferManager.InitBuffer(); for (var i = ; i < _backlog; i++)
{
var readWriteEventArg = new SocketAsyncEventArgs();
_bufferManager.SetBuffer(readWriteEventArg);
_readWritePool.Push(readWriteEventArg);
}
} public void StartServer()
{
try
{
Init();
IPAddress ipAddress = IPAddress.Any;
IPEndPoint localEndPoint = new IPEndPoint(ipAddress, _port);
_listenSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
_listenSocket.Bind(localEndPoint);
_listenSocket.Listen(_backlog);
StartAccept(null);
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
} private void StartAccept(SocketAsyncEventArgs acceptEventArg)
{
if (acceptEventArg == null)
{
acceptEventArg = new SocketAsyncEventArgs();
acceptEventArg.Completed += StartAccept_Completed;
}
else
{
acceptEventArg.AcceptSocket = null;
} _maxNumberAcceptedClients.WaitOne(); if (!_listenSocket.AcceptAsync(acceptEventArg))
{
ProcessAccept(acceptEventArg);
}
} private void StartAccept_Completed(object sender, SocketAsyncEventArgs e)
{
ProcessAccept(e);
} private void ProcessAccept(SocketAsyncEventArgs e)
{
if (e.SocketError == SocketError.Success)
{
var socket = e.AcceptSocket;
if (socket.Connected)
{
SocketAsyncEventArgs readEventArgs = _readWritePool.Pop();
readEventArgs.AcceptSocket = socket;
readEventArgs.Completed += ProcessAccept_Completed;
if (!socket.ReceiveAsync(readEventArgs))
{
ProcessReceiveFindFileRequest(readEventArgs);
}
StartAccept(e);
}
}
} private void ProcessAccept_Completed(object sender, SocketAsyncEventArgs e)
{
ProcessReceiveFindFileRequest(e);
} private void ProcessReceiveFindFileRequest(SocketAsyncEventArgs e)
{
var bytesRead = e.BytesTransferred;
if (bytesRead > && e.SocketError == SocketError.Success)
{
var receiveData = e.Buffer.Skip(e.Offset).Take(bytesRead).ToArray();
var dataList = PacketUtils.SplitBytes(receiveData, PacketUtils.ClientFindFileInfoTag());
if (dataList != null && dataList.Any())
{
var request = PacketUtils.GetData(PacketUtils.ClientFindFileInfoTag(), dataList.FirstOrDefault());
string str = System.Text.Encoding.UTF8.GetString(request);
var infos = str.Split('_');
var productName = infos[];
var revitVersion = infos[];
var currentVersion = infos[]; var mainFolder = AppDomain.CurrentDomain.BaseDirectory.Replace("bin", "TestFile");
var serverFileFolder = Path.Combine(mainFolder, "Server");
var serverFileFiles = new DirectoryInfo(serverFileFolder).GetFiles(); var updatefile = serverFileFiles.FirstOrDefault(x => x.Name.Contains(productName) && x.Name.Contains(revitVersion) && x.Name.Contains(currentVersion));
if (updatefile != null)
{
if (string.IsNullOrEmpty(updatefile.FullName) || !File.Exists(updatefile.FullName)) return;
_serverPath = updatefile.FullName; //ready to send back to Client
byte[] foundUpdateFileData = PacketUtils.PacketData(PacketUtils.ServerFoundFileInfoTag(), null); Array.Clear(e.Buffer, e.Offset, e.Count);
Array.Copy(foundUpdateFileData, , e.Buffer, e.Offset, foundUpdateFileData.Length); e.Completed -= ProcessAccept_Completed;
e.Completed += ProcessReceiveFindFileRequest_Completed; if (!e.AcceptSocket.SendAsync(e))
{
ProcessFilePosition(e);
}
}
}
}
} private void ProcessReceiveFindFileRequest_Completed(object sender, SocketAsyncEventArgs e)
{
ProcessFilePosition(e);
} private void ProcessFilePosition(SocketAsyncEventArgs e)
{
if (e.SocketError == SocketError.Success)
{
var socket = e.AcceptSocket;
if (socket.Connected)
{
//clear buffer
Array.Clear(e.Buffer, e.Offset, e.Count); e.Completed -= ProcessReceiveFindFileRequest_Completed;
e.Completed += ProcessFilePosition_Completed; if (!socket.ReceiveAsync(e))
{
ProcessSendFile(e);
}
}
}
} private void ProcessFilePosition_Completed(object sender, SocketAsyncEventArgs e)
{
ProcessSendFile(e);
} private void ProcessSendFile(SocketAsyncEventArgs e)
{
var bytesRead = e.BytesTransferred;
if (bytesRead > && e.SocketError == SocketError.Success)
{
var receiveData = e.Buffer.Skip(e.Offset).Take(bytesRead).ToArray();
var dataList = PacketUtils.SplitBytes(receiveData, PacketUtils.ClientRequestFileTag());
if (dataList != null)
{
foreach (var request in dataList)
{
if (PacketUtils.IsPacketComplete(request))
{
int startPosition = PacketUtils.GetRequestFileStartPosition(request); var packetSize = PacketUtils.GetPacketSize(_serverPath, _downloadChannelsCount);
if (packetSize != )
{
byte[] filedata = FileUtils.GetFile(_serverPath, startPosition, packetSize);
byte[] packetNumber = BitConverter.GetBytes(startPosition / packetSize); Console.WriteLine("Receive File Request PacketNumber: "+startPosition / packetSize); if (filedata != null)
{
//ready to send back to Client
byte[] segmentedFileResponseData = PacketUtils.PacketData(PacketUtils.ServerResponseFileTag(), filedata, packetNumber); Array.Clear(e.Buffer, e.Offset, e.Count);
Array.Copy(segmentedFileResponseData, , e.Buffer, e.Offset, segmentedFileResponseData.Length); e.Completed -= ProcessFilePosition_Completed;
e.Completed += ProcessSendFile_Completed; if (!e.AcceptSocket.SendAsync(e))
{
CloseClientSocket(e);
}
}
}
}
}
}
}
else
{
CloseClientSocket(e);
}
} private void ProcessSendFile_Completed(object sender, SocketAsyncEventArgs e)
{
CloseClientSocket(e);
} private void CloseClientSocket(SocketAsyncEventArgs e)
{
try
{
e.AcceptSocket.Shutdown(SocketShutdown.Both);
e.AcceptSocket.Close();
}
catch (Exception ex)
{
Console.WriteLine(ex.Message);
}
finally
{
_maxNumberAcceptedClients.Release();
_readWritePool.Push(e);
}
}
}
}
四、总结
坑坑洼洼总算是写完了SAEA的代码，由于本人知识面有限，如果说的不对，还请各位及时直接提出批评与建议，我这个人比较在乎技术不在乎面子的。
附：
MSDN示例：
https://msdn.microsoft.com/en-us/library/system.net.sockets.socketasynceventargs(v=vs.110).aspx
启蒙博客：
http://www.cnblogs.com/gaochundong/p/csharp_tcp_service_models.html
大神改造：
http://freshflower.iteye.com/blog/2285272
架构狂魔：
http://www.cnblogs.com/jiahuafu/archive/2013/01/05/2845631.html
我的GitHub