簡介Linux C的線程池

来源:https://www.cnblogs.com/liudw-0215/archive/2018/04/28/8968873.html
-Advertisement-
Play Games

前言：前面有篇博客已經介紹了線程、線程的信號量和互斥鎖，請參考博客：http://www.cnblogs.com/liudw-0215/p/8966645.html，接下來將介紹線程池。 一、理解 線程池能有效的處理多個線程的併發問題，避免大量的線程因為互相強占系統資源導致阻塞現象，能夠有效的降低頻 ...


　　前言：前面有篇博客已經介紹了線程、線程的信號量和互斥鎖，請參考博客：http://www.cnblogs.com/liudw-0215/p/8966645.html，接下來將介紹線程池。

　　一、理解

　　線程池能有效的處理多個線程的併發問題，避免大量的線程因為互相強占系統資源導致阻塞現象，能夠有效的降低頻繁創建和銷毀線程對性能所帶來的開銷。

　　大多數的網路伺服器，包括Web伺服器都具有一個特點，就是單位時間內必須處理數目巨大的連接請求，但是處理時間卻是比較短的。在傳統的多線程伺服器模型中是這樣實現的：一旦有個請求到達，就創建一個新的線程，由該線程執行任務，任務執行完畢之後，線程就退出。這就是"即時創建，即時銷毀"的策略。儘管與創建進程相比，創建線程的時間已經大大的縮短，但是如果提交給線程的任務是執行時間較短，而且執行次數非常頻繁，那麼伺服器就將處於一個不停的創建線程和銷毀線程的狀態。這筆開銷是不可忽略的，尤其是線程執行的時間非常非常短的情況。

　　線程池就是為瞭解決上述問題的，它的實現原理是這樣的：在應用程式啟動之後，就馬上創建一定數量的線程，放入空閑的隊列中。這些線程都是處於阻塞狀態，這些線程只占一點記憶體，不占用CPU。當任務到來後，線程池將選擇一個空閑的線程，將任務傳入此線程中運行。當所有的線程都處在處理任務的時候，線程池將自動創建一定的數量的新線程，用於處理更多的任務。執行任務完成之後線程並不退出，而是繼續線上程池中等待下一次任務。當大部分線程處於阻塞狀態時，線程池將自動銷毀一部分的線程，回收系統資源。

　　下麵是一個簡單線程池的實現，這個線程池的代碼是我參考網上的一個例子實現的，併進行了加工和修改。

　　二、示例

　　主要由三個文件組成：threadpool.h頭文件、threadpool.c源文件和mainpool.c組成。源碼中已有重要的註釋，就不加以分析了。

  • threadpool.h頭文件：
#include "my.h"
struct job
{
    void* (*callback_function)(void *arg);    //線程回調函數


    void *arg;                                //回調函數參數
    struct job *next;
};

struct threadpool
{
    int thread_num;                   //線程池中開啟線程的個數
    int queue_max_num;                //隊列中最大job的個數
    struct job *head;                 //指向job的頭指針
    struct job *tail;                 //指向job的尾指針
    pthread_t *pthreads;              //線程池中所有線程的pthread_t
    pthread_mutex_t mutex;            //互斥信號量
    pthread_cond_t queue_empty;       //隊列為空的條件變數
    pthread_cond_t queue_not_empty;   //隊列不為空的條件變數
    pthread_cond_t queue_not_full;    //隊列不為滿的條件變數
    int queue_cur_num;                //隊列當前的job個數
    int queue_close;                  //隊列是否已經關閉
    int pool_close;                   //線程池是否已經關閉
};

//================================================================================================
////函數名：                   threadpool_init
////函數描述：                 初始化線程池
////輸入：                    [in] thread_num     線程池開啟的線程個數
////                         [in] queue_max_num  隊列的最大job個數 
////輸出：                    無
////返回：                    成功：線程池地址 失敗：NULL
////================================================================================================
//struct threadpool* threadpool_init(int thread_num, int queue_max_num);
//
////================================================================================================
////函數名：                    threadpool_add_job
////函數描述：                  向線程池中添加任務
////輸入：                     [in] pool                  線程池地址
////                          [in] callback_function     回調函數
////                          [in] arg                     回調函數參數
////輸出：                     無
////返回：                     成功：0 失敗：-1
////================================================================================================
int threadpool_add_job(struct threadpool *pool, void* (*callback_function)(void *arg), void *arg);
//
////================================================================================================
////函數名：                    threadpool_destroy
////函數描述：                   銷毀線程池
////輸入：                      [in] pool                  線程池地址
////輸出：                      無
////返回：                      成功：0 失敗：-1
////================================================================================================
int threadpool_destroy(struct threadpool *pool);
//
////================================================================================================
////函數名：                    threadpool_function
////函數描述：                  線程池中線程函數
////輸入：                     [in] arg                  線程池地址
////輸出：                     無  
////返回：                     無
////================================================================================================
void* threadpool_function(void* arg);
  • 　　threadpool.c源文件： 
    #include "threadpool.h"
#include "my.h"

struct threadpool* threadpool_init(int thread_num, int queue_max_num)
{
    struct threadpool *pool = NULL;
    do 
    {
    pool = malloc(sizeof(struct threadpool));
    if (NULL == pool)
    {
        printf("failed to malloc threadpool!\n");
        break;
    }
    pool->thread_num = thread_num;
    pool->queue_max_num = queue_max_num;
    pool->queue_cur_num = 0;
    pool->head = NULL;
    pool->tail = NULL;
    if (pthread_mutex_init(&(pool->mutex), NULL))
    {
        printf("failed to init mutex!\n");
        break;
    }
    if (pthread_cond_init(&(pool->queue_empty), NULL))
    {
        printf("failed to init queue_empty!\n");
        break;
    }
    if (pthread_cond_init(&(pool->queue_not_empty), NULL))
    {
        printf("failed to init queue_not_empty!\n");
        break;
    }
    if (pthread_cond_init(&(pool->queue_not_full), NULL))
    {
        printf("failed to init queue_not_full!\n");
        break;
    }
    pool->pthreads = malloc(sizeof(pthread_t) * thread_num);
    if (NULL == pool->pthreads)
    {
        printf("failed to malloc pthreads!\n");
        break;
    }
    pool->queue_close = 0;
    pool->pool_close = 0;
    int i;
    for (i = 0; i < pool->thread_num; ++i)
    {
        pthread_create(&(pool->pthreads[i]), NULL, threadpool_function, (void *)pool);
    }

    return pool;    
    } while (0);

    return NULL;
}

int threadpool_add_job(struct threadpool* pool, void* (*callback_function)(void *arg), void *arg)
{
    assert(pool != NULL);
    assert(callback_function != NULL);
    assert(arg != NULL);

    pthread_mutex_lock(&(pool->mutex));
    while ((pool->queue_cur_num == pool->queue_max_num) && !(pool->queue_close || pool->pool_close))
    {
    pthread_cond_wait(&(pool->queue_not_full), &(pool->mutex));   //隊列滿的時候就等待
    }
    if (pool->queue_close || pool->pool_close)    //隊列關閉或者線程池關閉就退出
    {
    pthread_mutex_unlock(&(pool->mutex));
    return -1;
    }
    struct job *pjob =(struct job*) malloc(sizeof(struct job));
    if (NULL == pjob)
    {
    pthread_mutex_unlock(&(pool->mutex));
    return -1;
    } 
    pjob->callback_function = callback_function;    
    pjob->arg = arg;
    pjob->next = NULL;
    if (pool->head == NULL)   
    {
    pool->head = pool->tail = pjob;
    pthread_cond_broadcast(&(pool->queue_not_empty));  //隊列空的時候，有任務來時就通知線程池中的線程：隊列非空
    }
    else
    {
    pool->tail->next = pjob;
    pool->tail = pjob;    
    }
    pool->queue_cur_num++;
    pthread_mutex_unlock(&(pool->mutex));
    return 0;
}

void* threadpool_function(void* arg)
{
    struct threadpool *pool = (struct threadpool*)arg;
    struct job *pjob = NULL;
    while (1)  //死迴圈
    {
    pthread_mutex_lock(&(pool->mutex));
    while ((pool->queue_cur_num == 0) && !pool->pool_close)   //隊列為空時，就等待隊列非空
    {
        pthread_cond_wait(&(pool->queue_not_empty), &(pool->mutex));
    }
    if (pool->pool_close)   //線程池關閉，線程就退出
    {
        pthread_mutex_unlock(&(pool->mutex));
        pthread_exit(NULL);
    }
    pool->queue_cur_num--;
    pjob = pool->head;
    if (pool->queue_cur_num == 0)
    {
        pool->head = pool->tail = NULL;
    }
    else 
    {
        pool->head = pjob->next;
    }
    if (pool->queue_cur_num == 0)
    {
        pthread_cond_signal(&(pool->queue_empty));        //隊列為空，就可以通知threadpool_destroy函數，銷毀線程函數
    }
    if (pool->queue_cur_num == pool->queue_max_num - 1)
    {
        pthread_cond_broadcast(&(pool->queue_not_full));  //隊列非滿，就可以通知threadpool_add_job函數，添加新任務
    }
    pthread_mutex_unlock(&(pool->mutex));

    (*(pjob->callback_function))(pjob->arg);   //線程真正要做的工作，回調函數的調用
    free(pjob);
    pjob = NULL;    
    }
}
int threadpool_destroy(struct threadpool *pool)
{
    assert(pool != NULL);
    pthread_mutex_lock(&(pool->mutex));
    if (pool->queue_close || pool->pool_close)   //線程池已經退出了，就直接返回
    {
    pthread_mutex_unlock(&(pool->mutex));
    return -1;
    }

    pool->queue_close = 1;        //置隊列關閉標誌
    while (pool->queue_cur_num != 0)
    {
    pthread_cond_wait(&(pool->queue_empty), &(pool->mutex));  //等待隊列為空
    }    

    pool->pool_close = 1;      //置線程池關閉標誌
    pthread_mutex_unlock(&(pool->mutex));
    pthread_cond_broadcast(&(pool->queue_not_empty));  //喚醒線程池中正在阻塞的線程
    pthread_cond_broadcast(&(pool->queue_not_full));   //喚醒添加任務的threadpool_add_job函數
    int i;
    for (i = 0; i < pool->thread_num; ++i)
    {
    pthread_join(pool->pthreads[i], NULL);    //等待線程池的所有線程執行完畢
    }

    pthread_mutex_destroy(&(pool->mutex));          //清理資源
    pthread_cond_destroy(&(pool->queue_empty));
    pthread_cond_destroy(&(pool->queue_not_empty));   
    pthread_cond_destroy(&(pool->queue_not_full));    
    free(pool->pthreads);
    struct job *p;
    while (pool->head != NULL)
    {
    p = pool->head;
    pool->head = p->next;
    free(p);
    }
    free(pool);
    return 0;
}

    mainpool.c文件：

  • #include "threadpool.h"

void* work(void* arg)
{
    char *p = (char*) arg;
    printf("threadpool callback fuction : %s.\n", p);
    sleep(1);
}

int main(void)
{
    struct threadpool *pool = (struct threadpool *)threadpool_init(10, 20);
    threadpool_add_job(pool, work, "1");
    threadpool_add_job(pool, work, "2");
    threadpool_add_job(pool, work, "3");
    threadpool_add_job(pool, work, "4");
    threadpool_add_job(pool, work, "5");
    threadpool_add_job(pool, work, "6");
    threadpool_add_job(pool, work, "7");
    threadpool_add_job(pool, work, "8");
    threadpool_add_job(pool, work, "9");
    threadpool_add_job(pool, work, "10");
    threadpool_add_job(pool, work, "11");
    threadpool_add_job(pool, work, "12");
    threadpool_add_job(pool, work, "13");
    threadpool_add_job(pool, work, "14");
    threadpool_add_job(pool, work, "15");
    threadpool_add_job(pool, work, "16");
    threadpool_add_job(pool, work, "17");
    threadpool_add_job(pool, work, "18");
    threadpool_add_job(pool, work, "19");
    threadpool_add_job(pool, work, "20");
    threadpool_add_job(pool, work, "21");
    threadpool_add_job(pool, work, "22");
    threadpool_add_job(pool, work, "23");
    threadpool_add_job(pool, work, "24");
    threadpool_add_job(pool, work, "25");
    threadpool_add_job(pool, work, "26");
    threadpool_add_job(pool, work, "27");
    threadpool_add_job(pool, work, "28");
    threadpool_add_job(pool, work, "29");
    threadpool_add_job(pool, work, "30");
    threadpool_add_job(pool, work, "31");
    threadpool_add_job(pool, work, "32");
    threadpool_add_job(pool, work, "33");
    threadpool_add_job(pool, work, "34");
    threadpool_add_job(pool, work, "35");
    threadpool_add_job(pool, work, "36");
    threadpool_add_job(pool, work, "37");
    threadpool_add_job(pool, work, "38");
    threadpool_add_job(pool, work, "39");
    threadpool_add_job(pool, work, "40");

    sleep(5);
    threadpool_destroy(pool);
    return 0;
}

     

  •  


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • c# 數據寫入三菱PLC
    public void Data_Write_Value(int[] m_write_data) { int lSize = 2; string[] AddressArray = new string[18] { "D720", "D721", "D722", "D723", "D724", "D7 ...
  • 在Ubuntu16.04上配置.Net Core 2 環境
    一、安裝.Net Core SDK 按照官方文檔，執行以下命令安裝SDK curl https://packages.microsoft.com/keys/microsoft.asc | gpg --dearmor > microsoft.gpg sudo mv microsoft.gpg /etc ...
  • UGUI關卡選項界面
    1.Image組件—“Source Image”，Set Native Size. 2.Image組件—“Image Type”——Sliced 編輯要放大縮小的圖片，Sprite Editor，採用九宮格切圖。 3.創建空物體（作為父物體），添加Componment—Layout—Grid Lay ...
  • .Net Core 學習筆記1——包、元包、框架
    .Net Core 是由NuGet包（package）組成的平臺。一起使用的多個包的集合：元包（Metapackage）package 包 (對應以前的程式集概念)Framework 框架assembly 程式集重要的NuGet包：System.Runtime 基礎包System.Collectio... ...
  • Linux常用基本命令( tree, pwd, cd )
    pwd與cd命令 >pwd命令是“print working directory”中每個單詞的首字母縮寫，其功能是顯示當前工作目錄的絕對路徑。在實際工作中，我們在命令行操作命令時，經常會在各個目錄路徑之間進行切換，此時可使用pwd命令快速查看當前我們所在的目錄路徑。 還有一個 $PWD 環境變數，可 ...
  • FreeRTOS數據結構（一）--鏈表和鏈表項
    結構體定義 根據定義可以看出以下幾點： 1. 鏈表結構是一個雙向鏈表。 2. 每個鏈表都由鏈表和鏈表項組成。 3. 鏈表結構體本身有一個迷你鏈表項，用來標記鏈表的結尾。 鏈表和鏈表項初始化 鏈表的操作 鏈表的插入會根據List Item Value進行升序排列，其他並沒有什麼特殊的地方。初始化並插入 ...
  • Ubuntu 下 vi 輸入方向鍵會變成 ABCD 的解決方法
    Ubuntu 下 vi 輸入方向鍵會變成 ABCD ，這個很煩人，這裡給出解決辦法，親測可行 ...
  • Ubuntu出現apt-get: Package has no installation candidate問題
    今天在安裝 vim 的時候出現了 Package 'vim' has no installation candidate的問題 解決方法如下：# apt-get update# apt-get upgrade# apt-get install <packagename> 這樣就可以正常使用 apt- ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類