[C#]I/O完成端口的类定义和测试实例

从William Kennedy那里整理过来的,不同之处在于他自己定义了一个Overlapped，而我们这里直接使用
System.Threading.NativeOverlapped.

附一段我以前的Win32下的IOCP文档，如果您了解IOCP也可以直接跳过看后面的C#测试示范：

整理者：郑昀 @UltraPower

我们采用的是 I/O Complete Port （以下简称 IO CP ）处理机制。

简单的讲，当服务应用程序初始化时，它应该先创建一个 I/O CP 。我们在请求到来后 ，将得到的数据打包用 PostQueuedCompletionStatus 发送到 IOCP 中。这时需要创建一些个线程 (7 个线程 / 每 CPU ，再多就没有意义了 ) 来处理发送到 IOCP 端口的消息。实现步骤大致如下：

1 先在主线程中调用 CreateIoCompletionPort 创建 IOCP 。

CreateIoCompletionPort 的前三个参数只在把设备同 Complete Port 相关联时才有用。

此时我们只需传递 INVALID_HANDLE_VALUE,NULL 和 0 即可。

第四个参数告诉端口同时能运行的最多线程数，这里设置为 0 ，表示默认为当前计算机的 CPU 数目。

2 我们的 ThreadFun 线程函数执行一些初始化之后，将进入一个循环，该循环会在服务进程终止时才结束。

在循环中，调用 GetQueuedCompletionStatus ，这样就把当前线程的 ID 放入一个等待线程队列中， I/O CP 内核对象就总能知道哪个线程在等待处理完成的 I/O 请求。

如果在 IDLE_THREAD_TIMEOUT 规定的时间内 I/O CP 上还没有出现一个 Completion Packet ，则转入下一次循环。在这里我们设置的 IDLE_THREAD_TIMEOUT 为 1 秒。

当端口的 I/O 完成队列中出现一项时，完成端口就唤醒等待线程队列中的这个线程，该线程将得到完成的 I/O 项中的信息： 传输的字节数、完成键和 OVERLAPPED 结构的地址。

在我们的程序中可以用智能指针或者 BSTR 或者 int 来接受这个 OVERLAPPED 结构的地址的值，从而得到消息；然后在这个线程中处理消息。

GetQueuedCompletionStatus 的第一个参数 hCompletionPort 指出了要监视哪一个端口，这里我们传送先前从 CreateIoCompletionPort 返回的端口句柄。

需要注意的是：

第一， 线程池的数目是有限制的，和 CPU 数目有关系。

第二， IOCP 是一种较为完美的睡眠 / 唤醒 线程机制；线程当前没有任务要处理时，就进入睡眠状态，从而不占用 CPU 资源，直到被内核唤醒；

第三， 最近一次刚执行完的线程，下次任务来的时候还会唤醒它；所以有可能比较少被调用的线程以后被调用的几率也少。

测试代码：

using System;
using System.Threading; // Included for the Thread.Sleep call
using Continuum.Threading;
using System.Runtime.InteropServices;

namespace IOCPDemo
{
// =============================================================================
/**/ ///

1<summary> Sample class for the threading class  </summary>

public class UtilThreadingSample
{
// *****************************************************************************
/**/ ///

1<summary> Test Method  </summary>

static void Main()
{
// Create the MSSQL IOCP Thread Pool
IOCPThreadPool pThreadPool = new IOCPThreadPool( 0 , 10 , 20 , new IOCPThreadPool.USER_FUNCTION(IOCPThreadFunction));

// for(int i =1;i<10000;i++)
{
pThreadPool.PostEvent( 1234 );
}

Thread.Sleep( 100 );

pThreadPool.Dispose();
}

// ********************************************************************
/**/ ///

1<summary> Function to be called by the IOCP thread pool. Called when   
2![](http://www.cnblogs.com/Images/OutliningIndicators/InBlock.gif) ///  a command is posted for processing by the SocketManager  </summary>

///

1<param name="iValue"/>

The value provided by the thread posting the event
static public void IOCPThreadFunction( int iValue)
{
try
{
Console.WriteLine( " Value: {0} " , iValue.ToString());
Thread.Sleep( 3000 );
}

catch (Exception pException)
{
Console.WriteLine(pException.Message);
}
}
}

}

类代码：

using System;
using System.Threading;
using System.Runtime.InteropServices;

namespace IOCPThreading
{
[StructLayout(LayoutKind.Sequential, CharSet = CharSet.Auto)]

public sealed class IOCPThreadPool
{
[DllImport( " Kernel32 " , CharSet = CharSet.Auto)]
private unsafe static extern UInt32 CreateIoCompletionPort(UInt32 hFile, UInt32 hExistingCompletionPort, UInt32 * puiCompletionKey, UInt32 uiNumberOfConcurrentThreads);

[DllImport( " Kernel32 " , CharSet = CharSet.Auto)]
private unsafe static extern Boolean CloseHandle(UInt32 hObject);

[DllImport( " Kernel32 " , CharSet = CharSet.Auto)]
private unsafe static extern Boolean PostQueuedCompletionStatus(UInt32 hCompletionPort, UInt32 uiSizeOfArgument, UInt32 * puiUserArg, System.Threading.NativeOverlapped * pOverlapped);

[DllImport( " Kernel32 " , CharSet = CharSet.Auto)]
private unsafe static extern Boolean GetQueuedCompletionStatus(UInt32 hCompletionPort, UInt32 * pSizeOfArgument, UInt32 * puiUserArg, System.Threading.NativeOverlapped ** ppOverlapped, UInt32 uiMilliseconds);

private const UInt32 INVALID_HANDLE_VALUE = 0xffffffff ;
private const UInt32 INIFINITE = 0xffffffff ;
private const Int32 SHUTDOWN_IOCPTHREAD = 0x7fffffff ;
public delegate void USER_FUNCTION( int iValue);
private UInt32 m_hHandle;
private UInt32 GetHandle { get { return m_hHandle; } set { m_hHandle = value; } }

private Int32 m_uiMaxConcurrency;

private Int32 GetMaxConcurrency { get { return m_uiMaxConcurrency; } set { m_uiMaxConcurrency = value; } }


private Int32 m_iMinThreadsInPool;

private Int32 GetMinThreadsInPool { get { return m_iMinThreadsInPool; } set <span sty