SocketAsyncEventArgs是一个套接字操作的类，主要作用是实现socket消息的异步接收和发送，跟Socket的BeginSend和
BeginReceive方法异步处理没有多大区别，它的优势在于完成端口的实现来处理大数据的并发情况，由于本人学习不久，对千万级的
数据访问还没有多大体会，这里的简单实现作为一个学习的笔记，请酌情参考，如有错误，请及时指正。
先说说SockeAsyncEventArgs类的操作方法，以下是摘自MSDN的内容（cn/library/system.net.sockets.socketasynceventargs.aspx">MSDN的SockeAsyncEventArgs类描述）：
1、分配一个新的 SocketAsyncEventArgs 上下文对象，或者从应用程序池中获取一个空闲的此类对象。
SocketAsyncEventArgs saea = new SocketAsyncEventArgs();
//或者(这里的SocketAsyncEventArgsPool类一般是自己实现，MSDN有通过栈结构实现的程序池，也可以使用队列或链表)：
SocketAsyncEventArgs saea = new SocketAsyncEventArgsPool().Pop();
2、将该上下文对象的属性设置为要执行的操作（例如，完成回调方法、数据缓冲区、缓冲区偏移量以及要传输的最大数据量）。SocketAsyncEventArgs一般会根据操作执行相同的回调函数,所有设置内容在不同操作的回调都可以访问，我在调试时发现不能同时收发消息（可能是半双工），因此使用接收和发送试用两个对象
byte[] buffer = new byte[];
saea.SetBuffer(buffer, , buffer.Length); //设置缓冲区
saea.Completed += new EventHandler<SocketAsyncEventArgs>(MethodName); //设置回调方法
saea.RemoteEndPoint = new IPEndPoint(IPAddress.Any,); //设置远端连接节点，一般用于接收消息
saea.UserToken = new AsyncUserToken(); //设置用户信息，一般把连接的Socket对象放在这里
3、调用适当的套接字方法 (xxxAsync) 以启动异步操作。
4、如果异步套接字方法 (xxxAsync) 返回 true，则在回调中查询上下文属性来获取完成状态。
5、如果异步套接字方法 (xxxAsync) 返回 false，则说明操作是同步完成的。 可以查询上下文属性来获取操作结果。
//这是调用套接字的方法，即socket调用的方法：
Socket socket = saea.AcceptSocket;
socket.ConnectAsync(saea); //异步进行连接
socket.AcceptAsync(saea); //异步接收连接
socket.ReceiveAsync(saea); //异步接收消息
socket.SendAsync(saea); //异步发送消息
//这里注意的是，每个操作方法返回的是布尔值，这个布尔值的作用，是表明当前操作是否有等待I/O的情况，如果返回false则表示当前是同步操作，不需要等待，此时要要同步执行回调方法，一般写法是
bool willRaiseEvent = socket.ReceiveAsync(saea); //继续异步接收消息
if (!willRaiseEvent)
{
MethodName(saea);
}
6、将该上下文重用于另一个操作，将它放回到应用程序池中，或者将它丢弃。
如果用于持续监听连接，要注意saea.AcceptSocket = null;只有把saea对象的AcceptSocket置为null，才能监听到新的连接；
如果只用于单次通讯，则在用完saea对象是可丢弃，saea.Dispose(),如果想重复利用，则设置相应的异步操作即可，
saea.AcceptSocket = null;//重新监听
socket.ReceiveAsync(saea);//重新接收
socket.SendAsync(saea);//重新发送
关于具体的实现，类似于之前我记下的简单Socket通信，先是SocketServerManager的实现，该实现也是参考自MSDN：
public class SocketServerManager
{
readonly Socket _socket; //监听Socket
readonly EndPoint _endPoint;
private const int Backlog = ; //允许连接数目
private int byteSize = ; //同时UI界处理的事件
public delegate void OnEventCompletedHanlder(MessageFormat msg);
public event OnEventCompletedHanlder OnReceiveCompletedEvent;
public event OnEventCompletedHanlder OnSendCompletedEvent;
public event OnEventCompletedHanlder OnConnectedEvent;
public event OnEventCompletedHanlder OnDisconnectEvent;
public event OnEventCompletedHanlder OnNotConnectEvent; //private BufferManager bufferManager; //消息缓存管理
SocketAsyncEventArgsPool rwPool; //SAEA池
private Semaphore maxClient;
private Dictionary<string, SocketAsyncEventArgs> dicSAEA = null; public SocketServerManager(string ip, int port)
{
_socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
IPAddress ipAddress = IPAddress.Parse(ip);
_endPoint = new IPEndPoint(ipAddress, port);
//bufferManager = new BufferManager(totalBytes, byteSize);
maxClient = new Semaphore(Backlog, Backlog);
Init();
} public void Init()
{
//bufferManager.InitBuffer();
SocketAsyncEventArgs rwEventArgs;
rwPool = new SocketAsyncEventArgsPool(Backlog);
dicSAEA = new Dictionary<string, SocketAsyncEventArgs>();
for (int i = ; i < ; i++)
{
rwEventArgs = new SocketAsyncEventArgs();
rwEventArgs.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
rwEventArgs.UserToken = new AsyncUserToken(); rwEventArgs.SetBuffer(new byte[byteSize],,byteSize);
//bufferManager.SetBuffer(rwEventArgs); rwPool.Push(rwEventArgs);
}
} /// <summary>
/// 开启Socket监听
/// </summary>
public void Start()
{
_socket.Bind(_endPoint); //绑定本地地址进行监听
_socket.Listen(Backlog); //设置监听数量 StartAccept(null);
} public void StartAccept(SocketAsyncEventArgs acceptEventArg)
{
if (acceptEventArg == null)
{
acceptEventArg = new SocketAsyncEventArgs();
acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(OnConnectedCompleted);
}
else
{
acceptEventArg.AcceptSocket = null;
} maxClient.WaitOne();
bool willRaiseEvent = _socket.AcceptAsync(acceptEventArg);
if (!willRaiseEvent)
{
ProcessAccept(acceptEventArg);
}
} private void ProcessAccept(SocketAsyncEventArgs e)
{
if (e.SocketError != SocketError.Success) return; //异步处理失败，不做处理
SocketAsyncEventArgs saea = rwPool.Pop();
AsyncUserToken token = saea.UserToken as AsyncUserToken;
token.UserSocket = e.AcceptSocket; //获取远端对话Socket对象
string ipRemote = token.UserSocket.RemoteEndPoint.ToString();
string ip = token.UserSocket.RemoteEndPoint.ToString();
MessageFormat msg = new MessageFormat(string.Format("远程地址[{0}]成功连接到本地", ipRemote), ip, MsgType.Empty);
msg.tag = MsgType.Empty;
if (OnConnectedEvent != null) OnConnectedEvent(msg); //调用UI方法处理 //连接成功后，发送消息通知远程客户端
//OnSend("Connected Success !", _sendSocket.RemoteEndPoint);
SocketAsyncEventArgs sendArgs = new SocketAsyncEventArgs();
sendArgs.RemoteEndPoint = token.UserSocket.RemoteEndPoint;
sendArgs.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
sendArgs.UserToken = saea.UserToken;
dicSAEA.Add(token.UserSocket.RemoteEndPoint.ToString(), sendArgs);
bool willRaiseEvent = e.AcceptSocket.ReceiveAsync(saea);
if (!willRaiseEvent)
{
OnReceiveCompleted(saea);
} StartAccept(e);
} /// <summary>
/// 远端地址连接本地成功的回调
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
public void OnConnectedCompleted(object sender, SocketAsyncEventArgs e)
{
ProcessAccept(e);
} public void IO_Completed(object sender, SocketAsyncEventArgs e)
{
switch (e.LastOperation)
{
case SocketAsyncOperation.Receive:
OnReceiveCompleted(e);
break;
case SocketAsyncOperation.Send:
OnSendCompleted(e);
break;
default:
throw new ArgumentException("The last operation completed on the socket was not a receive or send");
}
} /// <summary>
/// 执行异步发送消息
/// </summary>
/// <param name="msg">消息内容</param>
/// <param name="ip">发送远端地址</param>
public void OnSend(MessageFormat mf)
{
if (!dicSAEA.ContainsKey(mf.ipStr))
{
if (OnNotConnectEvent != null)
{
OnNotConnectEvent(new MessageFormat("不存在此连接客户端","",MsgType.Empty));
return;
}
}
SocketAsyncEventArgs saea = dicSAEA[mf.ipStr];
AsyncUserToken token = saea.UserToken as AsyncUserToken;
if (saea == null) return;
//saea.SetBuffer(sendBuffer, 0, sendBuffer.Length); //设置SAEA的buffer消息内容
byte[] sendBuffer = Encoding.Unicode.GetBytes(string.Format("[length={0}]{1}", mf.msgStr.Length, mf.msgStr));
saea.SetBuffer(sendBuffer, , sendBuffer.Length);
bool willRaiseEvent = token.UserSocket.SendAsync(saea);
if (!willRaiseEvent)
{
OnSendCompleted(saea);
}
} /// <summary>
/// 发送消息回调处理
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
public void OnSendCompleted(SocketAsyncEventArgs e)
{
AsyncUserToken token = e.UserToken as AsyncUserToken;
byte[] sendBuffer = e.Buffer;
string msgStr = Encoding.Unicode.GetString(sendBuffer);
string ipAddress = token.UserSocket.RemoteEndPoint.ToString();
MessageFormat msg = new MessageFormat(msgStr, ipAddress, MsgType.Send);
if (OnSendCompletedEvent != null) OnSendCompletedEvent(msg); //调用UI方法处理
} /// <summary>
/// 接收消息回调处理
/// </summary>
/// <param name="e"></param>
public void OnReceiveCompleted(SocketAsyncEventArgs e)
{
if (e.SocketError != SocketError.Success) return; //判断消息的接收状态
AsyncUserToken token = e.UserToken as AsyncUserToken;
int lengthBuffer = e.BytesTransferred; //获取接收的字节长度
string ipAddress = token.UserSocket.RemoteEndPoint.ToString();
MessageFormat msg = new MessageFormat();
//如果接收的字节长度为0，则判断远端服务器关闭连接
if (lengthBuffer <= )
{
msg.msgStr = "远端服务器已经断开连接";
msg.ipStr = ipAddress;
msg.tag = MsgType.Handler;
if (OnDisconnectEvent != null) OnDisconnectEvent(msg);
CloseClientSocket(e);
}
else
{
byte[] receiveBuffer = e.Buffer;
byte[] buffer = new byte[lengthBuffer];
Buffer.BlockCopy(receiveBuffer, , buffer, , lengthBuffer);
msg.msgStr = Encoding.Unicode.GetString(buffer);
msg.ipStr = ipAddress;
msg.tag = MsgType.Receive;
bool willRaiseEvent = token.UserSocket.ReceiveAsync(e); //继续异步接收消息
if (!willRaiseEvent)
{
OnReceiveCompleted(e);
}
if (OnReceiveCompletedEvent != null) OnReceiveCompletedEvent(msg); //调用UI方法处理
}
} private void CloseClientSocket(SocketAsyncEventArgs e)
{
AsyncUserToken token = e.UserToken as AsyncUserToken;
try
{
token.UserSocket.Shutdown(SocketShutdown.Send);
}
catch (Exception) { }
dicSAEA.Remove(token.UserSocket.RemoteEndPoint.ToString());
token.UserSocket.Close(); maxClient.Release();
rwPool.Push(e);
}
}
public class SocketServerManager
通过一个栈结构SocketAsyncEventArgsPool保存的SocketAsyncEventArgs对象是用于接收消息，为了做到双方通讯，我每次在接收到远端客户端连接时，就创建一个新的SocketAsyncEventArgs对象保存在Dictionary结构中，这样在消息发送是就可以根据Ip来发送给远程客户端。
客户端的实现比较简单，不用考虑多方的通讯：
public class SocketClientManager
{
readonly Socket _socket; //用于消息交互的socket对象
readonly EndPoint _endPoint; //远端地址
readonly SocketAsyncEventArgs _saea; //处理连接和接收SAEA对象
//处理发送的SAEA处理，由于绑定不同的回调函数，因此需要不同的SAEA对象
SocketAsyncEventArgs _sendSaea; //处理UI的事件
public delegate void OnEventCompletedHanlder(MessageFormat msgFormat);
public event OnEventCompletedHanlder OnConnectedEvent; //连接成功事件
public event OnEventCompletedHanlder OnReceiveCompletedEvent; //收到消息事件
public event OnEventCompletedHanlder OnSendCompletedEvent; //发送成功事件 public SocketClientManager(string ip, int port)
{
_socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
IPAddress ipAddress = IPAddress.Parse(ip);
_endPoint = new IPEndPoint(ipAddress, port);
_saea = new SocketAsyncEventArgs {RemoteEndPoint = _endPoint};
} /// <summary>
/// 开启进行远程连接
/// </summary>
public void Start()
{
_saea.Completed += OnConnectedCompleted;
_socket.ConnectAsync(_saea); //进行异步连接 } /// <summary>
/// 连接成功的事件回调函数
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
public void OnConnectedCompleted(object sender, SocketAsyncEventArgs e)
{
if (e.SocketError != SocketError.Success) return;
Socket socket = sender as Socket;
string ipRemote = socket.RemoteEndPoint.ToString();
MessageFormat messageFormat = new MessageFormat(string.Format("连接服务器[{0}]成功！", ipRemote), socket.LocalEndPoint.ToString(), MsgType.Empty);
if (OnConnectedEvent != null) OnConnectedEvent(messageFormat); //开启新的接受消息异步操作事件
var receiveSaea = new SocketAsyncEventArgs();
var receiveBuffer = new byte[ * ];
receiveSaea.SetBuffer(receiveBuffer, , receiveBuffer.Length); //设置消息的缓冲区大小
receiveSaea.Completed += OnReceiveCompleted; //绑定回调事件
receiveSaea.RemoteEndPoint = _endPoint;
_socket.ReceiveAsync(receiveSaea);
} /// <summary>
/// 接受消息的回调函数
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
public void OnReceiveCompleted(object sender, SocketAsyncEventArgs e)
{
if (e.SocketError == SocketError.OperationAborted) return;
var socket = sender as Socket;
MessageFormat messageFormat = new MessageFormat();
if (e.SocketError == SocketError.Success &&e.BytesTransferred > )
{
string ipAddress = socket.RemoteEndPoint.ToString();
int lengthBuffer = e.BytesTransferred;
byte[] receiveBuffer = e.Buffer;
byte[] buffer = new byte[lengthBuffer];
Buffer.BlockCopy(receiveBuffer, , buffer, , lengthBuffer);
messageFormat.msgStr = Encoding.Unicode.GetString(buffer);
messageFormat.ipStr = ipAddress;
messageFormat.tag = MsgType.Receive;;
socket.ReceiveAsync(e);
}
else if (e.SocketError == SocketError.ConnectionReset && e.BytesTransferred == )
{
messageFormat.msgStr = "服务器已经断开连接";
messageFormat.ipStr = socket.RemoteEndPoint.ToString();
messageFormat.tag = MsgType.Handler;
}
else
{
return;
}
if (OnReceiveCompletedEvent != null) OnReceiveCompletedEvent(messageFormat);
} /// <summary>
/// 发送消息回调函数
/// </summary>
/// <param name="sender"></param>
/// <param name="e"></param>
public void OnSendCompleted(object sender, SocketAsyncEventArgs e)
{
if (e.SocketError != SocketError.Success) return;
var socket = sender as Socket;
byte[] sendBuffer = e.Buffer;
MessageFormat messageFormat = new MessageFormat(Encoding.Unicode.GetString(sendBuffer),socket.RemoteEndPoint.ToString(),MsgType.Send);
if (OnSendCompletedEvent != null) OnSendCompletedEvent(messageFormat);
} /// <summary>
/// 断开连接
/// </summary>
public void OnDisConnect()
{
if (_socket != null)
{
try
{
_socket.Shutdown(SocketShutdown.Both);
}
catch (SocketException ex)
{
}
finally
{
_socket.Close();
}
}
} /// <summary>
/// 发送消息
/// </summary>
/// <param name="msg"></param>
public void SendMsg(MessageFormat mf)
{
byte[] sendBuffer = Encoding.Unicode.GetBytes(string.Format("[length={0}]{1}", mf.msgStr.Length, mf.msgStr));
if (_sendSaea == null)
{
_sendSaea = new SocketAsyncEventArgs {RemoteEndPoint = _endPoint};
_sendSaea.Completed += OnSendCompleted;
}
_sendSaea.SetBuffer(sendBuffer, , sendBuffer.Length);
if (_socket != null) _socket.SendAsync(_sendSaea);
}
}
public class SocketClientManager
同样是使用收发不同的SocketAsyncEventArgs对象。
另外，关于解决缓存容量不足以容纳一条消息的半包问题，这里使用了简单的字符处理类，这个类是复制自Jimmy Zhang的类RequestHandler，当然，方法还是不完善的，难以解决不同顺序的消息接受。
具体源码（.net4.5，vs2013）：具体源码 http://files.cnblogs.com/files/supheart/ServerBySocket.zip