感謝原作者

重複造輪子的意義在於深刻的理解和學習---

main.cs:

/*using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;


namespace c2iocp
{
    class Program
    {
        static void Main(string[] args)
        {
        }
    }
}
*/
using System;
using System.Threading;  // Included for the Thread.Sleep call
using Continuum.Threading;
namespace Sample
{
    //============================================
    /// <summary> Sample class for the threading class </summary>
    public class UtilThreadingSample
    {
        //******************************************* 
        /// <summary> Test Method </summary>
        static void Main()
        {
            // Create the MSSQL IOCP Thread Pool
            IOCPThreadPool pThreadPool = new IOCPThreadPool(0, 5, 10, new IOCPThreadPool.USER_FUNCTION(IOCPThreadFunction));
            pThreadPool.PostEvent(10);
            Thread.Sleep(10000);
            pThreadPool.Dispose();
        }
        //*****************************************
        /// <summary> Function to be called by the IOCP thread pool.  Called when
        ///           a command is posted for processing by the SocketManager </summary>
        /// <param name="iValue"> The value provided by the thread posting the event </param>
        static public void IOCPThreadFunction(Int32 iValue)
        {
            try
            {
                Console.WriteLine("Value: {0}", iValue);
            }
            catch (Exception pException)
            {
                Console.WriteLine(pException.Message);
            }
        }
    }
}
 

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

namespace Continuum.Threading
{

    // Structures
    //==========================================
    /// <summary> This is the WIN32 OVERLAPPED structure </summary>
    [StructLayout(LayoutKind.Sequential, CharSet = CharSet.Auto)]
    public unsafe struct OVERLAPPED
    {
        UInt32* ulpInternal;
        UInt32* ulpInternalHigh;
        Int32 lOffset;
        Int32 lOffsetHigh;
        UInt32 hEvent;
    }

    // Classes
    //============================================
    /// <summary> This class provides the ability to create a thread pool to manage work.  The
    ///           class abstracts the Win32 IOCompletionPort API so it requires the use of
    ///           unmanaged code.  Unfortunately the .NET framework does not provide this functionality </summary>
    public sealed class IOCPThreadPool
    {

        // Win32 Function Prototypes
        /// <summary> Win32Func: Create an IO Completion Port Thread Pool </summary>
        [DllImport("Kernel32", CharSet = CharSet.Auto)]
        private unsafe static extern UInt32 CreateIoCompletionPort(UInt32 hFile, UInt32 hExistingCompletionPort, UInt32* puiCompletionKey, UInt32 uiNumberOfConcurrentThreads);

        /// <summary> Win32Func: Closes an IO Completion Port Thread Pool </summary>
        [DllImport("Kernel32", CharSet = CharSet.Auto)]
        private unsafe static extern Boolean CloseHandle(UInt32 hObject);

        /// <summary> Win32Func: Posts a context based event into an IO Completion Port Thread Pool </summary>
        [DllImport("Kernel32", CharSet = CharSet.Auto)]
        private unsafe static extern Boolean PostQueuedCompletionStatus(UInt32 hCompletionPort, UInt32 uiSizeOfArgument, UInt32* puiUserArg, OVERLAPPED* pOverlapped);

        /// <summary> Win32Func: Waits on a context based event from an IO Completion Port Thread Pool.
        ///           All threads in the pool wait in this Win32 Function </summary>
        [DllImport("Kernel32", CharSet = CharSet.Auto)]
        private unsafe static extern Boolean GetQueuedCompletionStatus(UInt32 hCompletionPort, UInt32* pSizeOfArgument, UInt32* puiUserArg, OVERLAPPED** ppOverlapped, UInt32 uiMilliseconds);

        // Constants
        /// <summary> SimTypeConst: This represents the Win32 Invalid Handle Value Macro </summary>
        private const UInt32 INVALID_HANDLE_VALUE = 0xffffffff;

        /// <summary> SimTypeConst: This represents the Win32 INFINITE Macro </summary>
        private const UInt32 INIFINITE = 0xffffffff;

        /// <summary> SimTypeConst: This tells the IOCP Function to shutdown </summary>
        private const Int32 SHUTDOWN_IOCPTHREAD = 0x7fffffff;



        // Delegate Function Types
        /// <summary> DelType: This is the type of user function to be supplied for the thread pool </summary>
        public delegate void USER_FUNCTION(Int32 iValue);


        // Private Properties
        private UInt32 m_hHandle;
        /// <summary> SimType: Contains the IO Completion Port Thread Pool handle for this instance </summary>
        private UInt32 GetHandle { get { return m_hHandle; } set { m_hHandle = value; } }

        private Int32 m_uiMaxConcurrency;
        /// <summary> SimType: The maximum number of threads that may be running at the same time </summary>
        private Int32 GetMaxConcurrency { get { return m_uiMaxConcurrency; } set { m_uiMaxConcurrency = value; } }

        private Int32 m_iMinThreadsInPool;
        /// <summary> SimType: The minimal number of threads the thread pool maintains </summary>
        private Int32 GetMinThreadsInPool { get { return m_iMinThreadsInPool; } set { m_iMinThreadsInPool = value; } }

        private Int32 m_iMaxThreadsInPool;
        /// <summary> SimType: The maximum number of threads the thread pool maintains </summary>
        private Int32 GetMaxThreadsInPool { get { return m_iMaxThreadsInPool; } set { m_iMaxThreadsInPool = value; } }

        private Object m_pCriticalSection;
        /// <summary> RefType: A serialization object to protect the class state </summary>
        private Object GetCriticalSection { get { return m_pCriticalSection; } set { m_pCriticalSection = value; } }

        private USER_FUNCTION m_pfnUserFunction;
        /// <summary> DelType: A reference to a user specified function to be call by the thread pool </summary>
        private USER_FUNCTION GetUserFunction { get { return m_pfnUserFunction; } set { m_pfnUserFunction = value; } }

        private Boolean m_bDisposeFlag;
        /// <summary> SimType: Flag to indicate if the class is disposing </summary>
        private Boolean IsDisposed { get { return m_bDisposeFlag; } set { m_bDisposeFlag = value; } }

        // Public Properties
        private Int32 m_iCurThreadsInPool;
        /// <summary> SimType: The current number of threads in the thread pool </summary>
        public Int32 GetCurThreadsInPool { get { return m_iCurThreadsInPool; } set { m_iCurThreadsInPool = value; } }
        /// <summary> SimType: Increment current number of threads in the thread pool </summary>
        private Int32 IncCurThreadsInPool() { return Interlocked.Increment(ref m_iCurThreadsInPool); }
        /// <summary> SimType: Decrement current number of threads in the thread pool </summary>
        private Int32 DecCurThreadsInPool() { return Interlocked.Decrement(ref m_iCurThreadsInPool); }
        private Int32 m_iActThreadsInPool;
        /// <summary> SimType: The current number of active threads in the thread pool </summary>
        public Int32 GetActThreadsInPool { get { return m_iActThreadsInPool; } set { m_iActThreadsInPool = value; } }
        /// <summary> SimType: Increment current number of active threads in the thread pool </summary>
        private Int32 IncActThreadsInPool() { return Interlocked.Increment(ref m_iActThreadsInPool); }
        /// <summary> SimType: Decrement current number of active threads in the thread pool </summary>
        private Int32 DecActThreadsInPool() { return Interlocked.Decrement(ref m_iActThreadsInPool); }
        private Int32 m_iCurWorkInPool;
        /// <summary> SimType: The current number of Work posted in the thread pool </summary>
        public Int32 GetCurWorkInPool { get { return m_iCurWorkInPool; } set { m_iCurWorkInPool = value; } }
        /// <summary> SimType: Increment current number of Work posted in the thread pool </summary>
        private Int32 IncCurWorkInPool() { return Interlocked.Increment(ref m_iCurWorkInPool); }
        /// <summary> SimType: Decrement current number of Work posted in the thread pool </summary>
        private Int32 DecCurWorkInPool() { return Interlocked.Decrement(ref m_iCurWorkInPool); }



        // Constructor, Finalize, and Dispose 
        //***********************************************
        /// <summary> Constructor </summary>
        /// <param name = "iMaxConcurrency"> SimType: Max number of running threads allowed </param>
        /// <param name = "iMinThreadsInPool"> SimType: Min number of threads in the pool </param>
        /// <param name = "iMaxThreadsInPool"> SimType: Max number of threads in the pool </param>
        /// <param name = "pfnUserFunction"> DelType: Reference to a function to call to perform work </param>
        /// <exception cref = "Exception"> Unhandled Exception </exception>
        public IOCPThreadPool(Int32 iMaxConcurrency, Int32 iMinThreadsInPool, Int32 iMaxThreadsInPool, USER_FUNCTION pfnUserFunction)
        {
            try
            {
                // Set initial class state
                GetMaxConcurrency = iMaxConcurrency;//系統允許的最大執行緒數，如果為0，則和核心數一致，一般都是0，
                GetMinThreadsInPool = iMinThreadsInPool;//最小執行緒數目
                GetMaxThreadsInPool = iMaxThreadsInPool;//最大執行緒數目，都是設定給的引數
                GetUserFunction = pfnUserFunction;//傳遞過來的回撥函式，或者說事件處理函式
                // Init the thread counters
                GetCurThreadsInPool = 0;//當前執行緒池裡的執行緒數
                GetActThreadsInPool = 0;//活動的執行緒數，正在工作的執行緒數
                GetCurWorkInPool = 0;//被投遞的事件數目
                // Initialize the Monitor Object
                GetCriticalSection = new Object();//來包含類狀態的物件？？？？
                // Set the disposing flag to false
                IsDisposed = false;//是否資源釋放標誌
                unsafe//c#呼叫外面dll的時候，需要用這個unsafe
                {
                    // Create an IO Completion Port for Thread Pool use
                    //步驟1：建立iocp埠,
                    /*
                     * 引數說明編輯
FileHandle是有效的檔案控制代碼或INVALID_HANDLE_VALUE。
ExistingCompletionPort是已經存在的完成埠。如果為NULL，則為新建一個IOCP。
CompletionKey是傳送給處理函式的引數。
NumberOfConcurrentThreads是有多少個執行緒在訪問這個訊息佇列。當引數ExistingCompletionPort不為0的時候，系統忽略該引數，當該引數為0表示允許同時相等數目於處理器個數的執行緒訪問該訊息佇列。
返回值編輯
返回一個IOCP的控制代碼。若為NULL則建立失敗，不為NULL則建立成功。
                     * 
                     * CreateIoCompletionPort的前三個引數只在把裝置同Complete Port相關聯時才有用。
此時我們只需傳遞INVALID_HANDLE_VALUE,NULL和0即可。
第四個引數告訴埠同時能執行的最多執行緒數，這裡設定為0，表示預設為當前計算機的CPU數目。
                     */
                    GetHandle = CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, null, (UInt32)GetMaxConcurrency);
                }
                // Test to make sure the IO Completion Port was created
                if (GetHandle == 0)
                    throw new Exception("Unable To Create IO Completion Port");
                // Allocate and start the Minimum number of threads specified
                Int32 iStartingCount = GetCurThreadsInPool;//當前執行緒池裡的執行緒數目
                ThreadStart tsThread = new ThreadStart(IOCPFunction);//// 摘要: 表示在 System.Threading.Thread 上執行的方法。public delegate void ThreadStart();
                /*
                 * 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返回的埠控制代碼。
                 */
                for (Int32 iThread = 0; iThread < GetMinThreadsInPool; ++iThread)
                {
                    // Create a thread and start it
                    Thread thThread = new Thread(tsThread);
                    thThread.Name = "IOCP " + thThread.GetHashCode();
                    thThread.Start();
                    // Increment the thread pool count
                    IncCurThreadsInPool();//原子操作+1初始為0
                    /*
                     * long m_dwRefCount..
InterlockedIncrement(&m_dwRefCount)
對m_dwRefCount加1
在對m_dwRefCount訪問的時候其他執行緒不能訪問這個變數
InterlockedIncrement實現的是原子性的加減，什麼是原子性的加減呢？

舉例：如果一個變數long value = 0；
首先說一下正常情況下的加減操作，value+=1；
1.系統從value的空間取出值，並動態生成一個空間儲存取出來的值；
2.將取出來的值和1做加法，並將和放入value的空間覆蓋掉原值，操作結束；
                    
如果此時有2個Thread，分別記作AThread，BThread
1.AThread將value從儲存空間取出，為0；
2.BThread將value從儲存空間取出，為0；
3.AThread將取出來的值和1做加法，並將和放入value的空間覆蓋掉原值，加法結束，value = 1；
4.BThread將取出來的值和1做加法，並將和放入value的空間覆蓋掉原值，加法結束，value = 1；
最後value應該是2，InterlockedIncrement保證一個執行緒訪問變數時其他執行緒不能訪問。
 
用於增減變數的並不是常用的Inc/Dec過程，而是用了InterlockedIncrement/InterlockedDecrement這一對過程，它們實現的功能完全一樣，都是對變數加一或減一。
但它們有一個最大的區別，那就是InterlockedIncrement/InterlockedDecrement是執行緒安全的。
即它們在多執行緒下能保證執行結果正確，而Inc/Dec不能。
或者按作業系統理論中的術語來說，這是一對“原語”操作。
                     */
                }
            }
            catch
            {
                throw new Exception("Unhandled Exception");
            }
        }

        //***********************************************
        /// <summary> Finalize called by the GC </summary>
        ~IOCPThreadPool()
        {
            /*
             * 這個類結束，應該也意味著是主執行緒了，所以只有主執行緒訪問？，既然是主執行緒，為什麼還要加標誌位呢？？？？
             */
            if (!IsDisposed)//還沒有釋放資源，則釋放，這個應該也是原子操作對isdisposeD吧
                Dispose();
        }

        //**********************************************
        /// <summary> Called when the object will be shutdown.  This
        ///           function will wait for all of the work to be completed
        ///           inside the queue before completing </summary>
        public void Dispose()
        {
            try
            {
                // Flag that we are disposing this object
                IsDisposed = true;
                // Get the current number of threads in the pool
                Int32 iCurThreadsInPool = GetCurThreadsInPool;//現在生成的執行緒數，用原子操作實現的
                // Shutdown all thread in the pool
                for (Int32 iThread = 0; iThread < iCurThreadsInPool; ++iThread)
                {
                    unsafe
                    {
                        bool bret = PostQueuedCompletionStatus(GetHandle, 4, (UInt32*)SHUTDOWN_IOCPTHREAD, null);
                        /*
                         * PostQueuedCompletionStatus函式，向每個工作者執行緒都發送—個特殊的完成資料包。該函式會指示每個執行緒都“立即結束並退出”.
                         * 下面是PostQueuedCompletionStatus函式的定義： 
[cpp] view plain copy
BOOL PostQueuedCompletionStatus(   
    HANDLE CompletlonPort,   
    DW0RD dwNumberOfBytesTrlansferred,   
    DWORD dwCompletlonKey,   
LPOVERLAPPED lpoverlapped,  
);  


       其中，CompletionPort引數指定想向其傳送一個完成資料包的完成埠物件。而就dwNumberOfBytesTransferred,dwCompletionKey和lpOverlapped
                         * 這三個引數來說.每—個都允許我們指定—個值,直接傳遞給GetQueuedCompletionStatus函式中對應的引數。這樣—來。
                         * —個工作者執行緒收到傳遞過來的三個GetQueuedCompletionStatus函式引數後，便可根據由這三個引數的某一個設定的特殊值，
                         * 決定何時應該退出。例如,可用dwCompletionPort引數傳遞0值,而—個工作者執行緒會將其解釋成中止指令。一旦所有工作者執行緒都已關閉,
                         * 便可使用CloseHandle函式,關閉完成埠。最終安全退出程式。 
        PostQueuedCompletionStatus函式提供了一種方式來與執行緒池中的所有執行緒進行通訊。如，當用戶終止服務應用程式時，我們想要所有執行緒都完全利索地退出。
                         * 但是如果各執行緒還在等待完成埠而又沒有已完成的I/O 請求，那麼它們將無法被喚醒。 
       通過為執行緒池中的每個執行緒都呼叫一次PostQueuedCompletionStatus，我們可以將它們都喚醒。每個執行緒會對GetQueuedCompletionStatus的返回值進行檢查，
                         * 如果發現應用程式正在終止，那麼它們就可以進行清理工作並正常地退出。
                         */
                    }
                }
                // Wait here until all the threads are gone
                while (GetCurThreadsInPool != 0) Thread.Sleep(100);
                unsafe
                {
                    // Close the IOCP Handle
                    CloseHandle(GetHandle);//便可使用CloseHandle函式,關閉完成埠。最終安全退出程式。 
                }
            }
            catch
            {
            }
        }

        // Private Methods
        //*******************************************
        /// <summary> IOCP Worker Function that calls the specified user function </summary>
        private void IOCPFunction()
        {
            UInt32 uiNumberOfBytes;
            Int32 iValue;
            try
            {
                while (true)
                {
                    unsafe
                    {
                        OVERLAPPED* pOv;
                        // Wait for an event
                        /*
                         * BOOL GetQueuedCompletionStatus(
HANDLE CompletionPort,
LPDWORD lpNumberOfBytes,
PULONG_PTR lpCompletionKey,
LPOVERLAPPED *lpOverlapped,
DWORD dwMilliseconds)；
呼叫引數：
CompletionPort：指定的IOCP，該值由CreateIoCompletionPort函式建立。
lpnumberofbytes：一次完成後的I/O操作所傳送資料的位元組數。
lpcompletionkey：當檔案I/O操作完成後，用於存放與之關聯的CK。
lpoverlapped：為呼叫IOCP機制所引用的OVERLAPPED結構。
dwmilliseconds：用於指定呼叫者等待CP的時間。
返回值：
呼叫成功，則返回非零數值，相關資料存於lpNumberOfBytes、lpCompletionKey、lpoverlapped變數中。失敗則返回零值。
                         */
                        GetQueuedCompletionStatus(GetHandle, &uiNumberOfBytes, (UInt32*)&iValue, &pOv, INIFINITE);
                    }
                    // Decrement the number of events in queue
                    DecCurWorkInPool();//原子操作，減少一個io裡面的任務數目
                    // Was this thread told to shutdown
                    if (iValue == SHUTDOWN_IOCPTHREAD)//關閉標誌位，post傳遞一個io完成資料，這裡讀到，知道該結束了，提前約定
                        break;
                    // Increment the number of active threads
                    IncActThreadsInPool();//這個執行緒讀到任務開始工作，增加活動執行緒數
                    try
                    {
                        // Call the user function
                        GetUserFunction(iValue);
                    }
                    catch
                    {
                    }
                    // Get a lock
                    Monitor.Enter(GetCriticalSection);
                    try
                    {
                        // If we have less than max threads currently in the pool
                        //當前已經生成了符合設定最小執行緒數目的執行緒，如果當前每到最大執行緒數，繼續建立
                        if (GetCurThreadsInPool < GetMaxThreadsInPool)
                        {
                            // Should we add a new thread to the pool
                            //所有執行緒都已在工作
                            if (GetActThreadsInPool == GetCurThreadsInPool)
                            {
                                if (IsDisposed == false)//沒有進行資源釋放，標誌位的作用在這裡
                                {
                                    // Create a thread and start it
                                    ThreadStart tsThread = new ThreadStart(IOCPFunction);
                                    Thread thThread = new Thread(tsThread);
                                    thThread.Name = "IOCP " + thThread.GetHashCode();
                                    thThread.Start();
                                    // Increment the thread pool count
                                    IncCurThreadsInPool();//增加當前的活動總數，子執行緒還可以繼續生成執行緒？？？？
                                }
                            }
                        }
                    }
                    catch
                    {
                    }
                    // Relase the lock
                    Monitor.Exit(GetCriticalSection);//這個是為了保證，比如當前程序為5都在工作了，最大上限設定10，
                    //發現都在工作只增加1個額外空閒執行緒，如果不用 Monitor 是不是會都發現不夠一下增加太多，最後可能當前執行緒數超過10
                    // Increment the number of active threads
                    DecActThreadsInPool();//幹活結束，減少忙碌狀態的執行緒
                }
            }
            catch
            {
            }
            // Decrement the thread pool count
            DecCurThreadsInPool();//執行緒退出，減少當前程序數
        }

        // Public Methods
        //******************************************
        /// <summary> IOCP Worker Function that calls the specified user function </summary>
        /// <param name="iValue"> SimType: A value to be passed with the event </param>
        /// <exception cref = "Exception"> Unhandled Exception </exception>
        public void PostEvent(Int32 iValue)
        {
            try
            {
                // Only add work if we are not disposing
                if (IsDisposed == false)
                {
                    unsafe
                    {
                        // Post an event into the IOCP Thread Pool
                        PostQueuedCompletionStatus(GetHandle, 4, (UInt32*)iValue, null);
                    }
                    // Increment the number of item of work
                    IncCurWorkInPool();
                    // Get a lock
                    Monitor.Enter(GetCriticalSection);
                    try
                    {
                        // If we have less than max threads currently in the pool
                        if (GetCurThreadsInPool < GetMaxThreadsInPool)
                        {
                            // Should we add a new thread to the pool
                            if (GetActThreadsInPool == GetCurThreadsInPool)
                            {
                                if (IsDisposed == false)
                                {
                                    // Create a thread and start it
                                    ThreadStart tsThread = new ThreadStart(IOCPFunction);
                                    Thread thThread = new Thread(tsThread);
                                    thThread.Name = "IOCP " + thThread.GetHashCode();
                                    thThread.Start();
                                    // Increment the thread pool count
                                    IncCurThreadsInPool();
                                }
                            }
                        }
                    }
                    catch
                    {
                    }
                    // Release the lock
                    Monitor.Exit(GetCriticalSection);
                }
            }
            catch (Exception e)
            {
                throw e;
            }
            catch
            {
                throw new Exception("Unhandled Exception");
            }
        }
        //*****************************************
        /// <summary> IOCP Worker Function that calls the specified user function </summary>
        /// <exception cref = "Exception"> Unhandled Exception </exception>
        public void PostEvent()
        {
            try
            {
                // Only add work if we are not disposing
                if (IsDisposed == false)
                {
                    unsafe
                    {
                        // Post an event into the IOCP Thread Pool
                        PostQueuedCompletionStatus(GetHandle, 0, null, null);
                    }
                    // Increment the number of item of work
                    IncCurWorkInPool();
                    // Get a lock
                    Monitor.Enter(GetCriticalSection);
                    try
                    {
                        // If we have less than max threads currently in the pool
                        if (GetCurThreadsInPool < GetMaxThreadsInPool)
                        {
                            // Should we add a new thread to the pool
                            if (GetActThreadsInPool == GetCurThreadsInPool)
                            {
                                if (IsDisposed == false)
                                {
                                    // Create a thread and start it
                                    ThreadStart tsThread = new ThreadStart(IOCPFunction);
                                    Thread thThread = new Thread(tsThread);
                                    thThread.Name = "IOCP " + thThread.GetHashCode();
                                    thThread.Start();
                                    // Increment the thread pool count
                                    IncCurThreadsInPool();
                                }
                            }
                        }
                    }
                    catch
                    {
                    }
                    // Release the lock
                    Monitor.Exit(GetCriticalSection);
                }
            }
            catch (Exception e)
            {
                throw e;
            }
            catch
            {
                throw new Exception("Unhandled Exception");
            }
        }
    }
}