快捷導(dǎo)航

C/C++ 原生API實(shí)現(xiàn)線程池的方法

更新時(shí)間：2021年11月01日 17:00:00 作者：lyshark

線程池，簡單來說就是有一堆已經(jīng)創(chuàng)建好的線程，接下來通過本文給大家介紹C/C++ 原生API實(shí)現(xiàn)線程池的方法，感興趣的朋友跟隨小編一起看看吧

線程池有兩個(gè)核心的概念，一個(gè)是任務(wù)隊(duì)列，一個(gè)是工作線程隊(duì)列。任務(wù)隊(duì)列負(fù)責(zé)存放主線程需要處理的任務(wù)，工作線程隊(duì)列其實(shí)是一個(gè)死循環(huán)，負(fù)責(zé)從任務(wù)隊(duì)列中取出和運(yùn)行任務(wù)，可以看成是一個(gè)生產(chǎn)者和多個(gè)消費(fèi)l者的模型。在一些高并發(fā)的網(wǎng)絡(luò)應(yīng)用中，線程池也是常用的技術(shù)。陳碩大神推薦的C++多線程服務(wù)端編程模式為：one loop per thread + thread pool，通常會(huì)有單獨(dú)的線程負(fù)責(zé)接受來自客戶端的請(qǐng)求，對(duì)請(qǐng)求稍作解析后將數(shù)據(jù)處理的任務(wù)提交到專門的計(jì)算線程池。

ThreadPool 線程池同步事件: 線程池內(nèi)的線程函數(shù)同樣支持互斥鎖,信號(hào)控制,內(nèi)核事件控制,臨界區(qū)控制.

#include <Windows.h>
#include <iostream>
#include <stdlib.h>

unsigned long g_count = 0;

// --------------------------------------------------------------
// 線程池同步-互斥量同步
void NTAPI TaskHandlerMutex(PTP_CALLBACK_INSTANCE Instance, PVOID Context, PTP_WORK Work)
{
	// 鎖定資源
	WaitForSingleObject(*(HANDLE *)Context, INFINITE);

	for (int x = 0; x < 100; x++)
	{
		printf("線程ID: %d ---> 子線程: %d \n", GetCurrentThreadId(), x);
		g_count = g_count + 1;
	}

	// 解鎖資源
	ReleaseMutexWhenCallbackReturns(Instance, *(HANDLE*)Context);
}

void TestMutex()
{
	// 創(chuàng)建互斥量
	HANDLE hMutex = CreateMutex(NULL, FALSE, NULL);

	PTP_WORK pool = CreateThreadpoolWork((PTP_WORK_CALLBACK)TaskHandlerMutex, &hMutex, NULL);

	for (int i = 0; i < 1000; i++)
	{
		SubmitThreadpoolWork(pool);
	}

	WaitForThreadpoolWorkCallbacks(pool, FALSE);
	CloseThreadpoolWork(pool);
	CloseHandle(hMutex);

	printf("相加后 ---> %d \n", g_count);
}

// --------------------------------------------------------------
// 線程池同步-事件內(nèi)核對(duì)象
void NTAPI TaskHandlerKern(PTP_CALLBACK_INSTANCE Instance, PVOID Context, PTP_WORK Work)
{
	// 鎖定資源
	WaitForSingleObject(*(HANDLE *)Context, INFINITE);

	for (int x = 0; x < 100; x++)
	{
		printf("線程ID: %d ---> 子線程: %d \n", GetCurrentThreadId(), x);
		g_count = g_count + 1;
	}

	// 解鎖資源
	SetEventWhenCallbackReturns(Instance, *(HANDLE*)Context);
}

void TestKern()
{
	HANDLE hEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
	SetEvent(hEvent);

	PTP_WORK pwk = CreateThreadpoolWork((PTP_WORK_CALLBACK)TaskHandlerKern, &hEvent, NULL);

	for (int i = 0; i < 1000; i++)
	{
		SubmitThreadpoolWork(pwk);
	}

	WaitForThreadpoolWorkCallbacks(pwk, FALSE);
	CloseThreadpoolWork(pwk);

	printf("相加后 ---> %d \n", g_count);
}

// --------------------------------------------------------------
// 線程池同步-信號(hào)量同步
void NTAPI TaskHandlerSemaphore(PTP_CALLBACK_INSTANCE Instance, PVOID Context, PTP_WORK Work)
{
	// 鎖定資源
	WaitForSingleObject(*(HANDLE *)Context, INFINITE);

	for (int x = 0; x < 100; x++)
	{
		printf("線程ID: %d ---> 子線程: %d \n", GetCurrentThreadId(), x);
		g_count = g_count + 1;
	}

	// 解鎖資源
	ReleaseSemaphoreWhenCallbackReturns(Instance, *(HANDLE*)Context, 1);
}

void TestSemaphore()
{
	// 創(chuàng)建信號(hào)量為100
	HANDLE hSemaphore = CreateSemaphore(NULL, 0, 100, NULL);

	ReleaseSemaphore(hSemaphore, 10, NULL);

	PTP_WORK pwk = CreateThreadpoolWork((PTP_WORK_CALLBACK)TaskHandlerSemaphore, &hSemaphore, NULL);

	for (int i = 0; i < 1000; i++)
	{
		SubmitThreadpoolWork(pwk);
	}

	WaitForThreadpoolWorkCallbacks(pwk, FALSE);
	CloseThreadpoolWork(pwk);
	CloseHandle(hSemaphore);

	printf("相加后 ---> %d \n", g_count);
}

// --------------------------------------------------------------
// 線程池同步-臨界區(qū)
void NTAPI TaskHandlerLeave(PTP_CALLBACK_INSTANCE Instance, PVOID Context, PTP_WORK Work)
{
	// 鎖定資源
	EnterCriticalSection((CRITICAL_SECTION*)Context);

	for (int x = 0; x < 100; x++)
	{
		printf("線程ID: %d ---> 子線程: %d \n", GetCurrentThreadId(), x);
		g_count = g_count + 1;
	}

	// 解鎖資源
	LeaveCriticalSectionWhenCallbackReturns(Instance, (CRITICAL_SECTION*)Context);
}

void TestLeave()
{
	CRITICAL_SECTION cs;
	InitializeCriticalSection(&cs);

	PTP_WORK pwk = CreateThreadpoolWork((PTP_WORK_CALLBACK)TaskHandlerLeave, &cs, NULL);

	for (int i = 0; i < 1000; i++)
	{
		SubmitThreadpoolWork(pwk);
	}

	WaitForThreadpoolWorkCallbacks(pwk, FALSE);
	DeleteCriticalSection(&cs);
	CloseThreadpoolWork(pwk);

	printf("相加后 ---> %d \n", g_count);
}

int main(int argc,char *argv)
{
	//TestMutex();
	//TestKern();
	//TestSemaphore();
	TestLeave();

	system("pause");
	return 0;
}

簡單的IO讀寫：

#include <Windows.h>
#include <iostream>
#include <stdlib.h>

// 簡單的異步文本讀寫
int ReadWriteIO()
{
	char enContent[] = "hello lyshark";
	char deContent[255] = { 0 };

	// 異步寫文件
	HANDLE hFileWrite = CreateFile(L"d://test.txt", GENERIC_WRITE, 0, NULL, OPEN_ALWAYS, FILE_FLAG_SEQUENTIAL_SCAN, NULL);
	if (INVALID_HANDLE_VALUE == hFileWrite)
	{
		return 0;
	}

	WriteFile(hFileWrite, enContent, strlen(enContent), NULL, NULL);
	FlushFileBuffers(hFileWrite);

	CancelSynchronousIo(hFileWrite);
	CloseHandle(hFileWrite);

	// 異步讀文件

	HANDLE hFileRead = CreateFile(L"d://test.txt", GENERIC_READ, 0, NULL, OPEN_ALWAYS, NULL, NULL);
	if (INVALID_HANDLE_VALUE == hFileRead)
	{
		return 0;
	}

	ReadFile(hFileRead, deContent, 255, NULL, NULL);
	CloseHandle(hFileRead);
	std::cout << "讀出內(nèi)容: " << deContent << std::endl;
	return 1;
}


// 通過IO獲取文件大小
int GetFileSize()
{
	HANDLE hFile = CreateFile(L"d://test.txt", 0, 0, NULL, OPEN_EXISTING, NULL, NULL);
	if (INVALID_HANDLE_VALUE == hFile)
	{
		return 0;
	}

	ULARGE_INTEGER ulFileSize;
	ulFileSize.LowPart = GetFileSize(hFile, &ulFileSize.HighPart);

	LARGE_INTEGER lFileSize;
	BOOL ret = GetFileSizeEx(hFile, &lFileSize);

	std::cout << "文件大小A: " << ulFileSize.QuadPart << " bytes" << std::endl;
	std::cout << "文件大小B: " << lFileSize.QuadPart << " bytes" << std::endl;
	CloseHandle(hFile);

	return 1;
}

// 通過IO設(shè)置文件指針和文件尾
int SetFilePointer()
{
	char deContent[255] = { 0 };
	DWORD readCount = 0;

	HANDLE hFile = CreateFile(L"d://test.txt", GENERIC_WRITE, 0, NULL, OPEN_ALWAYS, NULL, NULL);
	if (INVALID_HANDLE_VALUE == hFile)
	{
		return 0;
	}

	LARGE_INTEGER liMove;

	// 設(shè)置移動(dòng)位置
	liMove.QuadPart = 2;
	SetFilePointerEx(hFile, liMove, NULL, FILE_BEGIN);

	// 移動(dòng)到文件末尾
	SetEndOfFile(hFile);

	ReadFile(hFile, deContent, 255, &readCount, NULL);
	std::cout << "移動(dòng)指針后讀取: " << deContent << " 讀入長度: " << readCount << std::endl;

	CloseHandle(hFile);

	// 設(shè)置編碼格式
	_wsetlocale(LC_ALL, L"chs");
	setlocale(LC_ALL, "chs");
	wprintf(L"%s", deContent);
}

int main(int argc,char *argv)
{
	// 讀寫IO
	ReadWriteIO();

	// 取文件長度
	GetFileSize();

	// 設(shè)置文件指針
	SetFilePointer();

	return 0;
}

到此這篇關(guān)于C/C++ 原生API實(shí)現(xiàn)線程池的文章就介紹到這了,更多相關(guān)C++實(shí)現(xiàn)線程池內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: