一、 IOCP
IOCP（I/O Completion Port，I/O完成端口）是性能最好的一种I/O模型。它是应用程序使用线程池处理异步I/O请求的一种机制。在处理多个并发的异步I/O请求时，以往的模型都是在接收请求是创建一个线程来应答请求。这样就有很多的线程并行地运行在系统中。而这些线程都是可运行的，Windows内核花费大量的时间在进行线程的上下文切换，并没有多少时间花在线程运行上。再加上创建新线程的开销比较大，所以造成了效率的低下。
Windows Sockets应用程序在调用WSARecv()函数后立即返回,线程继续运行。当系统接收数据完成后，向完成端口发送通知包（这个过程对应用程序不可见）。
应用程序在发起接收数据操作后，在完成端口上等待操作结果。当接收到I/O操作完成的通知后，应用程序对数据进行处理。
image.png
完成端口其实就是上面两项的联合使用基础上进行了一定的改进。
Pool
/// <summary>
/// 与每个客户Socket相关联，进行Send和Receive投递时所需要的参数
/// </summary>
public class IoContextPool
{
List<SocketAsyncEventArgs> pool; //为每一个Socket客户端分配一个SocketAsyncEventArgs，用一个List管理，在程序启动时建立。
Int32 capacity; //pool对象池的容量
Int32 boundary; //已分配和未分配对象的边界，大的是已经分配的，小的是未分配的
public IoContextPool(Int32 capacity)
{
this.pool = new List<SocketAsyncEventArgs>(capacity);
this.boundary = 0;
this.capacity = capacity;
}
/// <summary>
/// 往pool对象池中增加新建立的对象，因为这个程序在启动时会建立好所有对象，
/// 故这个方法只在初始化时会被调用,因此，没有加锁。
/// </summary>
/// <param name="arg"></param>
/// <returns></returns>
public bool Add(SocketAsyncEventArgs arg)
{
if (arg != null && pool.Count < capacity)
{
pool.Add(arg);
boundary++;
return true;
}
else
return false;
}
/// <summary>
/// 取出集合中指定对象，内部使用
/// </summary>
/// <param name="index"></param>
/// <returns></returns>
//internal SocketAsyncEventArgs Get(int index)
//{
// if (index >= 0 && index < capacity)
// return pool[index];
// else
// return null;
//}
/// <summary>
/// 从对象池中取出一个对象，交给一个socket来进行投递请求操作
/// </summary>
/// <returns></returns>
public SocketAsyncEventArgs Pop()
{
lock (this.pool)
{
if (boundary > 0)
{
--boundary;
return pool[boundary];
}
else
return null;
}
}
/// <summary>
/// 一个socket客户断开，与其相关的IoContext被释放，重新投入Pool中，备用。
/// </summary>
/// <param name="arg"></param>
/// <returns></returns>
public bool Push(SocketAsyncEventArgs arg)
{
if (arg != null)
{
lock (this.pool)
{
int index = this.pool.IndexOf(arg, boundary); //找出被断开的客户,此处一定能查到，因此index不可能为-1，必定要大于0。
if (index == boundary) //正好是边界元素
boundary++;
else
{
this.pool[index] = this.pool[boundary]; //将断开客户移到边界上，边界右移
this.pool[boundary++] = arg;
}
}
return true;
}
else
return false;
}
}
Server
public partial class IocpServer : Form
{
private delegate void SetRichTextBoxCallBack(string str);
private SetRichTextBoxCallBack setRichTextBoxcallback;
public IocpServer()
{
setRichTextBoxcallback = new SetRichTextBoxCallBack(SetRichTextBoxReceive);
InitializeComponent();
}
/// <summary>
/// // 监听Socket，用于接受客户端的连接请求
/// </summary>
Socket Socketlistener;
/// <summary>
/// // 用于服务器执行的互斥同步对象
/// </summary>
private static Mutex mutex = new Mutex();
//完成端口上进行投递所用的IoContext对象池
//private IoContextPool ioContextPool;
//
/// <summary>
/// 服务器上连接的客户端总数
/// </summary>
private Int32 numConnectedSockets;
/// <summary>
/// 服务器能接受的最大连接数量
/// </summary>
private Int32 numConnections = 8192;
/// <summary>
/// 用于每个I/O Socket操作的缓冲区大小
/// </summary>
private Int32 bufferSize = 4028;
/// <summary>
/// 端口
/// </summary>
private Int32 bufferPort = Convert.ToInt32(ConfigurationManager.AppSettings["ServicePort"]);
//ip
private String _GetAddress = ConfigurationManager.AppSettings["ServiceAddress"];
//所有设备用户信息
//private List<Equipment> ListInfo = new List<Equipment>();
/// <summary>
/// 所有设备用户信息
/// </summary>
private List<ClientInformation> ListInfo = new List<ClientInformation>();
/// <summary>
/// 输出实体类
/// </summary>
HttpDate Hdate = new HttpDate();
//完成端口上进行投递所用的IoContext对象池
private IoContextPool ioContextPool;
DateTime GetDate;
TimeSpan UdpTime;
string FileTxt = Application.StartupPath + @"\FileTxt";
string FileName ;
private void IocpServer_Load(object sender, EventArgs e)
{
//获取所有设备用户信息
//ListInfo = AdoGetInfo.GetEquipmentUser();
GetDate = DateTime.Now.AddHours(-1);
FileSave();
this.numConnectedSockets = 0;
this.ioContextPool = new IoContextPool(numConnections);
// 为IoContextPool预分配SocketAsyncEventArgs对象
for (Int32 i = 0; i < this.numConnections; i++)
{
SocketAsyncEventArgs ioContext = new SocketAsyncEventArgs();
ioContext.Completed += new EventHandler<SocketAsyncEventArgs>(OnIOCompleted);
ioContext.SetBuffer(new Byte[this.bufferSize], 0, this.bufferSize);
// 将预分配的对象加入SocketAsyncEventArgs对象池中
this.ioContextPool.Add(ioContext);
}
// 获得主机相关信息
IPAddress[] addressList = Dns.GetHostEntry(Environment.MachineName).AddressList;
IPEndPoint localEndPoint = new IPEndPoint(addressList[addressList.Length - 1], bufferPort);
// 创建监听socket
this.Socketlistener = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
this.Socketlistener.ReceiveBufferSize = this.bufferSize;
this.Socketlistener.SendBufferSize = this.bufferSize;
if (localEndPoint.AddressFamily == AddressFamily.InterNetworkV6)
{
// 配置监听socket为 dual-mode (IPv4 & IPv6)
// 27 is equivalent to IPV6_V6ONLY socket option in the winsock snippet below,
this.Socketlistener.SetSocketOption(SocketOptionLevel.IPv6, (SocketOptionName)27, false);
this.Socketlistener.Bind(new IPEndPoint(IPAddress.IPv6Any, localEndPoint.Port));
}
else
{
this.Socketlistener.Bind(localEndPoint);
}
// 开始监听
this.Socketlistener.Listen(this.numConnections);
// 在监听Socket上投递一个接受请求。
this.StartAccept(null);
// Blocks the current thread to receive incoming messages.
mutex.WaitOne();
rTBoxInformation.Invoke(setRichTextBoxcallback, "服务器开始监听");
}
/// <summary>
/// 监听Socket接受处理
/// </summary>
/// <param name="e">SocketAsyncEventArg associated with the completed accept operation.</param>
private void ProcessAccept(SocketAsyncEventArgs e)
{
Socket s = e.AcceptSocket;
if (s.Connected)
{
try
{
SocketAsyncEventArgs ioContext = this.ioContextPool.Pop();
if (ioContext != null)
{
// 从接受的客户端连接中取数据配置ioContext
//ioContext.Completed += new EventHandler<SocketAsyncEventArgs>(OnIOCompleted);
//byte [] by=new Byte[]{};
//ioContext.SetBuffer(by, 0, by.Length);
//ioContext.UserToken = s;
// 从接受的客户端连接中取数据配置ioContext
ioContext.UserToken = s;
Interlocked.Increment(ref this.numConnectedSockets);
string outStr = String.Format("客户 {0} 连入, 共有 {1} 个连接。", s.RemoteEndPoint.ToString(), this.numConnectedSockets);
rTBoxInformation.Invoke(setRichTextBoxcallback, outStr);
if (!s.ReceiveAsync(ioContext))
{
this.ProcessReceive(ioContext);
}
}
else //已经达到最大客户连接数量，在这接受连接，发送“连接已经达到最大数”，然后断开连接
{
s.Send(Encoding.Default.GetBytes("连接已经达到最大数!"));
string outStr = String.Format("连接已满，拒绝 {0} 的连接。", s.RemoteEndPoint);
rTBoxInformation.Invoke(setRichTextBoxcallback, outStr);
s.Close();
}
}
catch (SocketException ex)
{
Socket token = e.UserToken as Socket;
string outStr = String.Format("接收客户 {0} 数据出错, 异常信息： {1} 。", token.RemoteEndPoint, ex.ToString());
AdoInsertTemp.AddServerErrorLog("接收客户数据出错:[IcopServer代码行号177]" + ex.Message);
rTBoxInformation.Invoke(setRichTextBoxcallback, outStr);
}
catch (Exception ex)
{
rTBoxInformation.Invoke(setRichTextBoxcallback, ex.Message);
AdoInsertTemp.AddServerErrorLog("监听Socket接受处理:[IcopServer代码行号182]" + ex.Message);
}
// 投递下一个接受请求
this.StartAccept(e);
}
}
/// <summary>
/// 从客户端开始接受一个连接操作
/// </summary>
/// <param name="acceptEventArg">The context object to use when issuing
/// the accept operation on the server's listening socket.</param>
private void StartAccept(SocketAsyncEventArgs acceptEventArg)
{
if (acceptEventArg == null)
{
acceptEventArg = new SocketAsyncEventArgs();
acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(OnAcceptCompleted);
}
else
{
// 重用前进行对象清理
acceptEventArg.AcceptSocket = null;
}
if (!this.Socketlistener.AcceptAsync(acceptEventArg))
{
this.ProcessAccept(acceptEventArg);
}
}
/// <summary>
///接收完成时处理函数
/// </summary>
/// <param name="e">与接收完成操作相关联的SocketAsyncEventArg对象</param>
private void ProcessReceive(SocketAsyncEventArgs e)
{
// 检查远程主机是否关闭连接
if (e.BytesTransferred > 0)
{
if (e.SocketError == SocketError.Success)
{
Socket s = (Socket)e.UserToken;
ClientInformation client = Hdate.AddClient(ListInfo, s.RemoteEndPoint.ToString(), e.BytesTransferred, e.Buffer);
//判断所有需接收的数据是否已经完成
if (s.Available == 0)
{
IPEndPoint localEp = s.RemoteEndPoint as IPEndPoint;
// 设置发送数据
byte[] _endRead = new byte[client.Transferred];
bool isClose = false;
client = Hdate.GetClient(ListInfo,client);
string strText = client.Rend; //Encoding.UTF8.GetString(e.Buffer, 0, client.Transferred);
RequestType REQUESTTYPE = Hdate.Request_Type(strText);//数据类型
RequestDeal REQUESTDEAL = Hdate.Request_Deal(strText);//命令方式
byte[] data = new byte[4028];
//初始化
if (REQUESTTYPE == RequestType.TypeGet && REQUESTDEAL == RequestDeal.GetConfiguration)
{
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("[来自{0}]{1}", localEp, strText));
data = CommonMethod.GetSend(bufferSize, Encoding.ASCII.GetBytes(Hdate.RtrunhttpNew(OutputPrint.ResponseGetFromNew.Replace("[GetSN]", client.DevicesSn))));
e.SetBuffer(data, e.Offset, data.Length);
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("向{0}发送：{1}", localEp, Encoding.UTF8.GetString(data)));
}
else if (REQUESTTYPE == RequestType.TypeGet && REQUESTDEAL == RequestDeal.GetInfo)
{
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("[来自{0}]{1}", localEp, strText));
data = CommonMethod.GetSend(bufferSize, Encoding.ASCII.GetBytes(Hdate.RtrunhttpNew(OutputPrint.capsOk)));
e.SetBuffer(data, e.Offset, data.Length);
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("向{0}发送：{1}", localEp, Encoding.UTF8.GetString(data)));
}
//是否有命令发送
else if (REQUESTTYPE == RequestType.TypeGet && REQUESTDEAL == RequestDeal.GetOrders)
{
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("[来自{0}]{1}", localEp, strText));
if (client.WaitingName != null)
{
if (client.WaitingName.Count > 0)
{
data = CommonMethod.GetSend(bufferSize, Encoding.ASCII.GetBytes(Hdate.RtrunhttpNew(client.WaitingName[0])));
}
else
{
data = CommonMethod.GetSend(bufferSize, Encoding.ASCII.GetBytes(Hdate.RtrunhttpNew(OutputPrint.capsOk)));
}
}
else
{
data = CommonMethod.GetSend(bufferSize, Encoding.ASCII.GetBytes(Hdate.RtrunhttpNew(OutputPrint.capsOk)));
}
e.SetBuffer(data, e.Offset, data.Length);
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("向{0}发送：{1}", localEp, Encoding.UTF8.GetString(data)));
}
//返回值说明：0 命令执行成功-1 参数错误-3 存取错误
else if (REQUESTTYPE == RequestType.TypePOST && REQUESTDEAL == RequestDeal.PostInfo)
{
Hdate.RemoveLength(client, strText);
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("[来自{0}]{1}", localEp, strText));
data = CommonMethod.GetSend(bufferSize, Encoding.ASCII.GetBytes(Hdate.RtrunhttpNew(OutputPrint.capsOk)));
e.SetBuffer(data, e.Offset, data.Length);
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("向{0}发送：{1}", localEp, Encoding.UTF8.GetString(data)));
}
//post发送数据命令
else if (REQUESTTYPE == RequestType.TypePOST && REQUESTDEAL == RequestDeal.PostAttTable)
{
Hdate.GetTable(strText, client);
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("[来自{0}]{1}", localEp, strText));
data = CommonMethod.GetSend(bufferSize, Encoding.ASCII.GetBytes(Hdate.RtrunhttpNew(OutputPrint.capsOk)));
e.SetBuffer(data, e.Offset, data.Length);
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("向{0}发送：{1}", localEp, Encoding.UTF8.GetString(data)));
}
else
{
if (!client.IsData)
{
Hdate.RemoveLength(client, strText);
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("[来自{0}]{1}", localEp, strText));
data = CommonMethod.GetSend(bufferSize, Encoding.ASCII.GetBytes(Hdate.RtrunhttpNew(OutputPrint.capsOk)));
e.SetBuffer(data, e.Offset, data.Length);
rTBoxInformation.Invoke(setRichTextBoxcallback, string.Format("向{0}发送：{1}", localEp, Encoding.UTF8.GetString(data)));
}
else
{
Hdate.GetTable(strText, client);
}
}
try
{
if (!s.SendAsync(e)) //投递发送请求，这个函数有可能同步发送出去，这时返回false，并且不会引发SocketAsyncEventArgs.Completed事件
{
// 同步发送时处理发送完成事件
this.ProcessSend(e, isClose);
}
if (client != null && !client.IsData)
{
Hdate.AddAtt(client, ListInfo);
Thread.Sleep(6000);
this.CloseClientSocket(s, e);
}
}
catch (Exception ex)
{
AdoInsertTemp.AddServerErrorLog("接收完成时处理函数:[IcopServer代码行号330]" + ex.Message);
}
}
else if (!s.ReceiveAsync(e)) //为接收下一段数据，投递接收请求，这个函数有可能同步完成，这时返回false，并且不会引发SocketAsyncEventArgs.Completed事件
{
// 同步接收时处理接收完成事件
this.ProcessReceive(e);
}
}
else
{
this.ProcessError(e);
}
}
else
{
this.CloseClientSocket(e);
}
}
/// <summary>
/// 发送完成时处理函数
/// </summary>
/// <param name="e">与发送完成操作相关联的SocketAsyncEventArg对象</param>
private void ProcessSend(SocketAsyncEventArgs e, bool isReceive)
{
try
{
if (e.SocketError == SocketError.Success)
{
Socket s = (Socket)e.UserToken;
if (s != null)
{
//this.CloseClientSocket(s, e);
//接收时根据接收的字节数收缩了缓冲区的大小，因此投递接收请求时，恢复缓冲区大小
//e.SetBuffer(new Byte[buffer_Size], 0, buffer_Size);
e.SetBuffer(0, bufferSize);
if (!s.ReceiveAsync(e)) //投递接收请求
{
// 同步接收时处理接收完成事件
this.ProcessReceive(e);
}
}
}
else
{
this.ProcessError(e);
}
}
catch (Exception ex)
{
rTBoxInformation.Invoke(setRichTextBoxcallback, ex.Message);
AdoInsertTemp.AddServerErrorLog("发送完成时处理函数:[IcopServer代码行号390]" + ex.Message);
this.ProcessError(e);
}
}
/// <summary>
/// 当Socket上的发送或接收请求被完成时，调用此函数
/// </summary>
/// <param name="sender">激发事件的对象</param>
/// <param name="e">与发送或接收完成操作相关联的SocketAsyncEventArg对象</param>
private void OnIOCompleted(object sender, SocketAsyncEventArgs e)
{
// Determine which type of operation just completed and call the associated handler.
switch (e.LastOperation)
{
case SocketAsyncOperation.Receive:
this.ProcessReceive(e);
break;
case SocketAsyncOperation.Send:
this.ProcessSend(e,true);
break;
default:
throw new ArgumentException("The last operation completed on the socket was not a receive or send");
}
}
/// <summary>
/// 处理socket错误
/// </summary>
/// <param name="e"></param>
private void ProcessError(SocketAsyncEventArgs e)
{
try
{
Socket s = e.UserToken as Socket;
IPEndPoint localEp = s.LocalEndPoint as IPEndPoint;
this.CloseClientSocket(s, e);
string outStr = String.Format("套接字错误 {0}, IP {1}, 操作 {2}。", (Int32)e.SocketError, localEp, e.LastOperation);
rTBoxInformation.Invoke(setRichTextBoxcallback, outStr);
}
catch (Exception ex) { AdoInsertTemp.AddServerErrorLog("处理socket错误:[IcopServer代码行号431]" + ex.Message); }
}
/// <summary>
/// 关闭socket连接
/// </summary>
/// <param name="e">SocketAsyncEventArg associated with the completed send/receive operation.</param>
private void CloseClientSocket(SocketAsyncEventArgs e)
{
Socket s = e.UserToken as Socket;
this.CloseClientSocket(s, e);
}
/// <summary>
/// accept 操作完成时回调函数
/// </summary>
/// <param name="sender">Object who raised the event.</param>
/// <param name="e">SocketAsyncEventArg associated with the completed accept operation.</param>
private void OnAcceptCompleted(object sender, SocketAsyncEventArgs e)
{
this.ProcessAccept(e);
}
private void CloseClientSocket(Socket s, SocketAsyncEventArgs e)
{
try
{
if (s != null && this.numConnectedSockets > 0)
{
Interlocked.Decrement(ref this.numConnectedSockets);
// SocketAsyncEventArg 对象被释放，压入可重用队列。
this.ioContextPool.Push(e);
string outStr = String.Format("客户 {0} 断开, 共有 {1} 个连接。", s.RemoteEndPoint.ToString(), this.numConnectedSockets);
rTBoxInformation.Invoke(setRichTextBoxcallback, outStr);
try
{
s.Shutdown(SocketShutdown.Send);
s.Disconnect(true);
}
catch (Exception ex)
{
rTBoxInformation.Invoke(setRichTextBoxcallback, ex.Message);
AdoInsertTemp.AddServerErrorLog("sokect关闭:[IcopServer代码行号477]" + ex.Message);
}
finally
{
s.Close();
}
}
}
catch (Exception ex) { AdoInsertTemp.AddServerErrorLog("sokect关闭:[IcopServer代码行号467]" + ex.Message); }
}
private void SetRichTextBoxReceive(string str)
{
//show txt
rTBoxInformation.AppendText(str);
//do right
rTBoxInformation.Select(this.rTBoxInformation.TextLength, 0);
//do down
rTBoxInformation.ScrollToCaret();
//new row
rTBoxInformation.AppendText("\r\n");
FileSave();
}
private void FileSave()
{
TimeSpan UdpTime=DateTime.Now-GetDate;
if(UdpTime.Hours>=1)
{
FileStream fs = null;
StreamWriter sw = null;
FileName = FileTxt + DateTime.Now.ToString("yyyyMMddHH");
if(!File.Exists(FileName))
{
Directory.CreateDirectory(FileName);
}
fs = new FileStream(FileName + @"\log_"+DateTime.Now.ToString("yyyyMMddHHmmssfff") + ".txt", FileMode.Create);
sw = new StreamWriter(fs);
sw.Write(rTBoxInformation.Text);
sw.Close();
fs.Close();
rTBoxInformation.Clear();
GetDate = DateTime.Now;
}
}
private void IocpServer_FormClosing(object sender, FormClosingEventArgs e)
{
e.Cancel = true;
this.Hide();
}
private void notifyIcon_MouseDoubleClick(object sender, MouseEventArgs e)
{
this.Show();
WindowState = FormWindowState.Normal;
}
}