线程池函数1 - 异步调用函数

本次主要针对默认线程池。

//1)使用TrySubmitThreadpoolCallback
//注意每次调用TrySubmitThreadpoolCallback，
//系统都会分配一个工作项work item。
//所以如果打算提交大量的工作项，出于性能和内存的考虑，
//最好使用下面另一组函数。/*
调用TrySubmitThreadpoolCallback，系统会自动分配work item，该函数通过PostQueuedCompletionStatus将该work item添加到线程池队列中，线程池也由系统创建，并让线程池中的一个线程来调用我们的回调函数，当这个线程处理完一个客户请求之后，它并不会立刻被销毁，而是回到线程池，准备好处理队列中的任何其它的工作项。
*///默认线程池，不用考虑后两个参数，定制时才需要
BOOL WINAPI TrySubmitThreadpoolCallback(_In_        PTP_SIMPLE_CALLBACK  pfns,_Inout_opt_ PVOID                pv,_In_opt_    PTP_CALLBACK_ENVIRON pcbe
);
//默认线程池，不用考虑参数，定制时才需要
VOID CALLBACK SimpleCallback(_Inout_     PTP_CALLBACK_INSTANCE Instance,_Inout_opt_ PVOID                 Context
);
//2）先用CreateThreadpoolWork创建一个工作项，
//然后使用同一个work item来调用SubmitThreadpoolWork提交多个请求
PTP_WORK WINAPI CreateThreadpoolWork(_In_        PTP_WORK_CALLBACK    pfnwk,_Inout_opt_ PVOID                pv,_In_opt_    PTP_CALLBACK_ENVIRON pcbe
);
VOID CALLBACK WorkCallback(_Inout_     PTP_CALLBACK_INSTANCE Instance,_Inout_opt_ PVOID                 Context,_Inout_     PTP_WORK              Work
);
VOID WINAPI SubmitThreadpoolWork(_Inout_ PTP_WORK pwk
);
VOID WINAPI WaitForThreadpoolWorkCallbacks(_Inout_ PTP_WORK pwk,_In_    BOOL     fCancelPendingCallbacks
);
VOID WINAPI CloseThreadpoolWork(_Inout_ PTP_WORK pwk
);

1.创建work object, 所有task共用

PTP_WORK WINAPI CreateThreadpoolWork(_In_        PTP_WORK_CALLBACK    pfnwk, //回调函数_Inout_opt_ PVOID                pv,    //传给回调函数的参数_In_opt_    PTP_CALLBACK_ENVIRON pcbe   //回调函数执行环境
);

2.定义异步执行的回调函数

Applications implement this callback if they call the SubmitThreadpoolWork
function to start a worker thread for the work object.

VOID CALLBACK WorkCallback(_Inout_     PTP_CALLBACK_INSTANCE Instance,_Inout_opt_ PVOID                 Context,_Inout_     PTP_WORK              Work
);

3.向线程池提交work请求

系统会自动创建一个默认的线程池，并让线程池中的一个线程来调用我们的回调函数
Posts a work object to the thread pool. A worker thread calls the work object’s callback function.

VOID WINAPI SubmitThreadpoolWork(_Inout_ PTP_WORK pwk //created by CreateThreadpoolWork
);

4.等待未完成的callback完成，同时可以取消pending的callback被执行

Waits for outstanding work callbacks to complete and optionally
cancels pending callbacks that have not yet started to execute.

VOID WINAPI WaitForThreadpoolWorkCallbacks(
_Inout_ PTP_WORK pwk,
_In_    BOOL     fCancelPendingCallbacks
);

5.释放work object

Releases the specified work object.
The work object is freed immediately if there are no outstanding callbacks;
otherwise, the work object is freed asynchronously after the outstanding callbacks complete.
If there is a cleanup group associated with the work object, it is not necessary to call this function;
calling the CloseThreadpoolCleanupGroupMembers function releases the work, wait, and timer objects associated with the cleanup group.

VOID WINAPI CloseThreadpoolWork(_Inout_ PTP_WORK pwk //created by CreateThreadpoolWork
);

6.实例1

下面的代码是按上面的2）来设计的

BatchDlg.h
class CBatchDlg : public CDialogEx
{
... ...
protected:virtual BOOL OnInitDialog();afx_msg void OnBnClickedOk();afx_msg void OnDestroy();afx_msg LRESULT OnCompleted(WPARAM wParam, LPARAM lParam);private:BOOL CreateWorkObject();BOOL SubmitWorkToThreadPool();void WaitCloseThreadPool();static void CALLBACK TaskHandler(PTP_CALLBACK_INSTANCE Instance, PVOID Context, PTP_WORK Work);private:CListBox m_oLstBox;PTP_WORK m_pWorkItem;
};BatchDlg.cpp
#include <strsafe.h>
#include <Windowsx.h>
#define WM_APP_COMPLETED (WM_APP+123)
HWND g_hDlg = nullptr;
volatile LONG g_nCurrentTask = 0;void AddMessage(LPCTSTR szMsg)
{HWND hListBox = GetDlgItem(g_hDlg, IDC_LIST1);ListBox_SetCurSel(hListBox, ListBox_AddString(hListBox, szMsg));
}
BOOL CBatchDlg::OnInitDialog()
{... ...g_hDlg = m_hWnd;BOOL bRet = CreateWorkObject();ASSERT(bRet);
}
void CBatchDlg::OnBnClickedOk()
{BOOL bRet = SubmitWorkToThreadPool();ASSERT(bRet);
}
LRESULT CBatchDlg::OnCompleted(WPARAM wParam, LPARAM lParam)
{TCHAR szMsg[MAX_PATH + 1];StringCchPrintf( szMsg, _countof(szMsg),TEXT("____Task #%u was the last task of the batch____"), lParam);AddMessage(szMsg);::Button_Enable(::GetDlgItem(m_hWnd, IDOK), TRUE);return 0;
}//-------------------关键代码如下-------------------
//1.创建work object
BOOL CBatchDlg::CreateWorkObject()
{ASSERT(nullptr == m_pWorkItem);// Create the work item that will be used by all tasksm_pWorkItem = CreateThreadpoolWork(TaskHandler, NULL, NULL);if (m_pWorkItem == nullptr) {::MessageBox(NULL, TEXT("Impossible to create the work item for tasks."), TEXT(""), MB_ICONSTOP);return FALSE;}return TRUE;
}
//2.定义异步执行的回调函数
void CALLBACK CBatchDlg::TaskHandler(PTP_CALLBACK_INSTANCE Instance, PVOID Context, PTP_WORK Work)
{LONG currentTask = InterlockedIncrement(&g_nCurrentTask);TCHAR szMsg[MAX_PATH];StringCchPrintf(szMsg, _countof(szMsg),TEXT("[%u] Task #%u is starting."), GetCurrentThreadId(), currentTask);AddMessage(szMsg);// Simulate a lot of workSleep(currentTask * 1000);StringCchPrintf(szMsg, _countof(szMsg),TEXT("[%u] Task #%u is done."), GetCurrentThreadId(), currentTask);AddMessage(szMsg);if (InterlockedDecrement(&g_nCurrentTask) == 0){   // Notify the UI thread for completion.::PostMessage(g_hDlg, WM_APP_COMPLETED, 0, (LPARAM)currentTask);}
}
//3.向线程池提交work请求
BOOL CBatchDlg::SubmitWorkToThreadPool()
{if (m_pWorkItem == nullptr) {::MessageBox(NULL, TEXT("Impossible to create the work item for tasks."), TEXT(""), MB_ICONSTOP);return FALSE;}::Button_Enable(::GetDlgItem(g_hDlg, IDOK), FALSE);AddMessage(TEXT("----Start a new batch----"));// Submit 4 tasks by using the same work itemSubmitThreadpoolWork(m_pWorkItem);SubmitThreadpoolWork(m_pWorkItem);SubmitThreadpoolWork(m_pWorkItem);SubmitThreadpoolWork(m_pWorkItem);AddMessage(TEXT("4 tasks are submitted."));return TRUE;
}
//4,5.释放work object
void CBatchDlg::WaitCloseThreadPool()
{if (m_pWorkItem == nullptr)return;//WaitForThreadpoolWorkCallbacks(m_pWorkItem, TRUE);CloseThreadpoolWork(m_pWorkItem);
}

7.实例2

7.1 MyObj .h/cpp，CMyObj 类，线程池函数要处理的对象

class CMyObj : public CObject
{
public:CMyObj(const CString& strName, int nNum, CListBox& lstBox);virtual ~CMyObj();public:BOOL DoSth();BOOL IsDone()const;protected:void AddMsg(const CString& strMsg);protected:CListBox&           m_lstBox;CString             m_strName;int                 m_nNum; BOOL                m_bIsDone;
};
// CMyObj
CMyObj::CMyObj(const CString& strName, int nNum, CListBox& lstBox): m_nNum(nNum), m_lstBox(lstBox), m_strName(strName), m_bIsDone(FALSE)
{
}
CMyObj::~CMyObj()
{TRACE(_T("[%s] is released\n"), m_strName);
}// CMyObj member functions
BOOL CMyObj::DoSth()
{CString strMsg;for (int i = 0; i < m_nNum; ++i){strMsg.Format(_T("%s, %04d, %p, threadId=%u"), m_strName, i, this, GetCurrentThreadId());AddMsg(strMsg);Sleep(500);}m_bIsDone = TRUE;return TRUE;
}BOOL CMyObj::IsDone()const
{return m_bIsDone;
}void CMyObj::AddMsg(const CString& strMsg)
{m_lstBox.AddString(strMsg);
}

7.2 ObjManager.h/cpp, CObjManager管理CMyObj对象

class CMyObj;
class CObjManager
{
public:CObjManager();~CObjManager();    //cleanvoid CleanCompleteObj();void ReleaseAll();//create objectvoid AddObjToBacklogList(const CString& strName, int nNum, CListBox& lstbox);void SetCreating(BOOL bCreate);BOOL IsCreating()const;//handle objectvoid SetRunning(BOOL bRun);BOOL IsRunning()const;BOOL HandleOneObj();void AddObjToWorkingList(CObject* obj);private:CObList             m_oBacklogList;CCriticalSection    m_csBacklogList;BOOL                m_bIsCreating;//CObList             m_oWorkingList;CCriticalSection    m_csWorkingList;BOOL                m_bIsRunning;
};
CObjManager::CObjManager(): m_bIsRunning(FALSE), m_bIsCreating(FALSE)
{
}
CObjManager::~CObjManager()
{ReleaseAll();
}//控制是否继续往链表里面添加待处理对象
void CObjManager::SetCreating(BOOL bCreate)
{m_bIsCreating = bCreate;
}
BOOL CObjManager::IsCreating()const
{return m_bIsCreating;
}//控制是否继续处理链表中的待处理项
void CObjManager::SetRunning(BOOL bRun)
{m_bIsRunning = bRun;
}
BOOL CObjManager::IsRunning()const
{return m_bIsRunning;
}//新增加一个待处理对象，放入backlog列表
void CObjManager::AddObjToBacklogList(const CString& strName, int nNum, CListBox& lstbox)
{CSingleLock sLock(&m_csBacklogList, TRUE);CMyObj* pObj = new CMyObj(strName, nNum, lstbox);if (nullptr != pObj){POSITION pos = m_oBacklogList.AddTail(pObj);ASSERT(pos);}
}//从backlog列表中拿一个对象，放入working列表，并开始处理这个对象
BOOL CObjManager::HandleOneObj()
{POSITION pos = nullptr;CMyObj* pObj = nullptr;{CSingleLock sLock(&m_csBacklogList, TRUE);pos = m_oBacklogList.GetHeadPosition();if (pos == nullptr)return FALSE;pObj = (CMyObj*)m_oBacklogList.GetAt(pos);m_oBacklogList.RemoveAt(pos);AddObjToWorkingList(pObj);}if (nullptr != pObj)pObj->DoSth();return TRUE;
}//把对象放入working列表
void CObjManager::AddObjToWorkingList(CObject* obj)
{if (nullptr == obj)return;CSingleLock sLock(&m_csWorkingList, TRUE);POSITION pos = m_oWorkingList.AddTail(obj);ASSERT(pos);
}void CObjManager::CleanCompleteObj()
{CMyObj* pObj = nullptr;POSITION pos = nullptr, posPre = nullptr;CSingleLock oLock(&m_csWorkingList, TRUE);for (pos = m_oWorkingList.GetHeadPosition();  (posPre = pos) != nullptr;){if (!IsRunning())break;pObj = (CMyObj*)m_oWorkingList.GetNext(pos);if (nullptr == pObj){m_oWorkingList.RemoveAt(posPre);            }   else if (pObj->IsDone()){m_oWorkingList.RemoveAt(posPre);delete pObj;}}
}void CObjManager::ReleaseAll()
{CObject* pObj = nullptr;POSITION pos = nullptr;{CSingleLock sLock(&m_csBacklogList, TRUE);for (pos = m_oBacklogList.GetHeadPosition(); pos != nullptr;){pObj = m_oBacklogList.GetNext(pos);if (nullptr == pObj)continue;delete pObj;}m_oBacklogList.RemoveAll();}//{CSingleLock oLock(&m_csWorkingList, TRUE);for (pos = m_oWorkingList.GetHeadPosition(); pos != nullptr;){pObj = m_oWorkingList.GetNext(pos);if (nullptr == pObj)continue;delete pObj;}m_oWorkingList.RemoveAll();}
}

7.3 对话框类

// CBatchDlg dialog
class CBatchDlg : public CDialogEx
{
... ...
protected:virtual BOOL OnInitDialog();afx_msg void OnDestroy();afx_msg void OnBnClickedOk();   afx_msg void OnBnClickedBtnHandleObj();
public:void CloseCreatObjThread();void CloseCleanObjThread();
private:BOOL CreateWorkObject();void SubmitWorkToThreadPool();void WaitCloseThreadPool();static void CALLBACK TaskHandler(PTP_CALLBACK_INSTANCE Instance, PVOID Context, PTP_WORK Work);static DWORD WINAPI CreateObjThread(LPVOID lpParam);static DWORD WINAPI CleanObjThread(LPVOID lpParam);public:CListBox            m_oLstBox;//输出信息static CObjManager  m_objMgr; //管理所有要处理的对象
private:PTP_WORK            m_pWorkItem;//线程池对象HANDLE              m_hCreatObjThread;//创建要处理对象的线程HANDLE              m_hCleanObjThread;//清理完成处理的对象
};CObjManager CBatchDlg::m_objMgr;CBatchDlg::CBatchDlg(CWnd* pParent /*=NULL*/): CDialogEx(IDD_BATCH_DIALOG, pParent), m_pWorkItem(nullptr), m_hCreatObjThread(nullptr), m_hCleanObjThread(nullptr)
{m_hIcon = AfxGetApp()->LoadIcon(IDR_MAINFRAME);
}
BEGIN_MESSAGE_MAP(CBatchDlg, CDialogEx)ON_WM_DESTROY()ON_BN_CLICKED(IDOK, &CBatchDlg::OnBnClickedOk)ON_BN_CLICKED(IDC_BTN_HANDLE_OBJ, &CBatchDlg::OnBnClickedBtnHandleObj)
END_MESSAGE_MAP()//1.定义线程池的线程函数（异步执行的回调函数）
void CALLBACK CBatchDlg::TaskHandler(PTP_CALLBACK_INSTANCE Instance, PVOID Context, PTP_WORK Work)
{while (m_objMgr.IsRunning()){if (!m_objMgr.HandleOneObj())Sleep(10);}TRACE(_T("Exit Thread CBatchDlg::TaskHandler: id = %x\n"), GetCurrentThreadId());
}//2.创建work object
BOOL CBatchDlg::OnInitDialog()
{CDialogEx::OnInitDialog();SetIcon(m_hIcon, TRUE);SetIcon(m_hIcon, FALSE);//2.创建work objectBOOL bRet = CreateWorkObject();ASSERT(bRet);return TRUE;  // return TRUE  unless you set the focus to a control
}
//2.创建work object
BOOL CBatchDlg::CreateWorkObject()
{ASSERT(nullptr == m_pWorkItem);// Create the work item that will be used by all tasksm_pWorkItem = CreateThreadpoolWork(TaskHandler, NULL, NULL);if (m_pWorkItem == nullptr) {::MessageBox(NULL, TEXT("Impossible to create the work item for tasks."), TEXT(""), MB_ICONSTOP);return FALSE;}return TRUE;
}//3.创建“生产者”线程，不断的创建待处理对象
void CBatchDlg::OnBnClickedOk()
{if (m_hCreatObjThread){m_objMgr.SetCreating(FALSE);SetDlgItemText(IDOK, _T("Start Create"));return;}m_objMgr.SetCreating(TRUE);m_hCreatObjThread = CreateThread(NULL,                   // default security attributes0,                      // use default stack size  CreateObjThread,       // thread function namethis,                  // argument to thread function 0,                      // use default creation flags NULL);                  // returns the thread identifier if (NULL == m_hCreatObjThread){m_objMgr.SetCreating(FALSE);SetDlgItemText(IDOK, _T("Start Create"));}else{SetDlgItemText(IDOK, _T("Stop Create"));}
}
DWORD WINAPI CBatchDlg::CreateObjThread(LPVOID lpParam)
{CBatchDlg* pThis = (CBatchDlg*)lpParam;if (nullptr == pThis)return 0;static LONG lCount = 0;CString strName(_T(""));while (m_objMgr.IsCreating()){++lCount;strName.Format(_T("Obj[%ld]"), lCount);m_objMgr.AddObjToBacklogList(strName, 10, pThis->m_oLstBox);Sleep(1000);}pThis->CloseCreatObjThread();TRACE(_T("Exit Thread CBatchDlg::CreateObjThread\n"));return 0;
}
void CBatchDlg::CloseCreatObjThread()
{if (nullptr != m_hCreatObjThread){CloseHandle(m_hCreatObjThread);m_hCreatObjThread = nullptr;}
}//4.创建清理线程，向线程池提交work请求
void CBatchDlg::OnBnClickedBtnHandleObj()
{if (nullptr == m_hCleanObjThread) {m_hCleanObjThread = CreateThread(NULL,                   // default security attributes0,                      // use default stack size  CleanObjThread,       // thread function nameNULL,                  // argument to thread function 0,                      // use default creation flags NULL);                  // returns the thread identifier }if (m_objMgr.IsRunning()){m_objMgr.SetRunning(FALSE);Sleep(3000);//等待线程池中的线程退出SetDlgItemText(IDC_BTN_HANDLE_OBJ, _T("Start Work"));return;}m_objMgr.SetRunning(TRUE);SubmitWorkToThreadPool();SetDlgItemText(IDC_BTN_HANDLE_OBJ, _T("Stop Work"));
}
void CBatchDlg::SubmitWorkToThreadPool()
{SubmitThreadpoolWork(m_pWorkItem);SubmitThreadpoolWork(m_pWorkItem);SubmitThreadpoolWork(m_pWorkItem);SubmitThreadpoolWork(m_pWorkItem);SubmitThreadpoolWork(m_pWorkItem);
}
DWORD WINAPI CBatchDlg::CleanObjThread(LPVOID lpParam)
{while (m_objMgr.IsRunning()){m_objMgr.CleanCompleteObj();Sleep(1000);}TRACE(_T("Exit Thread CBatchDlg::CleanObjThread\n"));return 0;
}
void CBatchDlg::CloseCleanObjThread()
{if (nullptr != m_hCleanObjThread){CloseHandle(m_hCleanObjThread);m_hCleanObjThread = nullptr;}
}//5. 释放“生产者”线程，等待线程池线程结束，释放线程池work object
void CBatchDlg::OnDestroy()
{m_objMgr.SetRunning(FALSE);m_objMgr.SetCreating(FALSE);Sleep(3000);CloseCreatObjThread();CloseCleanObjThread();WaitCloseThreadPool();CDialogEx::OnDestroy();
}
//5.释放work object
void CBatchDlg::WaitCloseThreadPool()
{if (m_pWorkItem == nullptr)return;//WaitForThreadpoolWorkCallbacks(m_pWorkItem, FALSE);//WaitForThreadpoolWorkCallbacks(m_pWorkItem, TRUE);CloseThreadpoolWork(m_pWorkItem);
}

8. 待解决问题 WaitForThreadpoolWorkCallbacks 导致失去响应

据说 devenv.exe /safemode运行就不会出这个现象，但试了不行。