我們一起手擼一個線程池

作者：程序喵大人 2021-10-27 06:49:34

見AddThread()函數，默認會創建Core線程，也可以選擇創建Cache線程，線程內部會有一個死循環，不停的等待任務，有任務到來時就會執行，同時內部會判斷是否是Cache線程，如果是Cache線程，timeout時間內沒有任務執行就會自動退出循環，線程結束。

[[431420]]

本文轉載自微信公眾號「程序喵大人」，作者程序喵大人。轉載本文請聯系程序喵大人公眾號。

之前分享過一次手寫線程池 - C語言版，然后有朋友問是否有C++線程池實現的文章：

其實關于C++線程池的文章我好久以前寫過，但估計很多新朋友都沒有看到過，這里也重新發一下!

本人在開發過程中經常會遇到需要使用線程池的需求，但查了一圈發現在C++中完備的線程池第三方庫還是比較少的，于是打算自己搞一個，鏈接地址文章最后附上，目前還只是初版，可能還有很多問題，望各位指正。

線程池都需要什么功能?

個人認為線程池需要支持以下幾個基本功能：

核心線程數(core_threads)：線程池中擁有的最少線程個數，初始化時就會創建好的線程，常駐于線程池。
最大線程個數(max_threads)：線程池中擁有的最大線程個數，max_threads>=core_threads，當任務的個數太多線程池執行不過來時，內部就會創建更多的線程用于執行更多的任務，內部線程數不會超過max_threads，多創建出來的線程在一段時間內沒有執行任務則會自動被回收掉，最終線程個數保持在核心線程數。
超時時間(time_out)：如上所述，多創建出來的線程在time_out時間內沒有執行任務就會被回收。
可獲取當前線程池中線程的總個數。
可獲取當前線程池中空閑線程的個數。
開啟線程池功能的開關。
關閉線程池功能的開關，可以選擇是否立即關閉，立即關閉線程池時，當前線程池里緩存的任務不會被執行。

如何實現線程池?下面是自己實現的線程池邏輯。

線程池中主要的數據結構

1. 鏈表或者數組：用于存儲線程池中的線程。

2. 隊列：用于存儲需要放入線程池中執行的任務。

3. 條件變量：當有任務需要執行時，用于通知正在等待的線程從任務隊列中取出任務執行。

代碼如下：

class ThreadPool { 
 public: 
  using PoolSeconds = std::chrono::seconds; 
 
  /** 線程池的配置 
   * core_threads: 核心線程個數，線程池中最少擁有的線程個數，初始化就會創建好的線程，常駐于線程池 
   * 
   * max_threads: >=core_threads，當任務的個數太多線程池執行不過來時， 
   * 內部就會創建更多的線程用于執行更多的任務，內部線程數不會超過max_threads 
   * 
   * max_task_size: 內部允許存儲的最大任務個數，暫時沒有使用 
   * 
   * time_out: Cache線程的超時時間，Cache線程指的是max_threads-core_threads的線程, 
   * 當time_out時間內沒有執行任務，此線程就會被自動回收 
   */ 
  struct ThreadPoolConfig { 
      int core_threads; 
      int max_threads; 
      int max_task_size; 
      PoolSeconds time_out; 
  }; 
 
  /** 
   * 線程的狀態：有等待、運行、停止 
   */ 
  enum class ThreadState { kInit = 0, kWaiting = 1, kRunning = 2, kStop = 3 }; 
 
  /** 
   * 線程的種類標識：標志該線程是核心線程還是Cache線程，Cache是內部為了執行更多任務臨時創建出來的 
   */ 
  enum class ThreadFlag { kInit = 0, kCore = 1, kCache = 2 }; 
 
  using ThreadPtr = std::shared_ptr<std::thread>; 
  using ThreadId = std::atomic<int>; 
  using ThreadStateAtomic = std::atomic<ThreadState>; 
  using ThreadFlagAtomic = std::atomic<ThreadFlag>; 
 
  /** 
   * 線程池中線程存在的基本單位，每個線程都有個自定義的ID，有線程種類標識和狀態 
   */ 
  struct ThreadWrapper { 
      ThreadPtr ptr; 
      ThreadId id; 
      ThreadFlagAtomic flag; 
      ThreadStateAtomic state; 
 
      ThreadWrapper() { 
          ptr = nullptr; 
          id = 0; 
          state.store(ThreadState::kInit); 
      } 
  }; 
  using ThreadWrapperPtr = std::shared_ptr<ThreadWrapper>; 
  using ThreadPoolLock = std::unique_lock<std::mutex>; 
 
 private: 
  ThreadPoolConfig config_; 
 
  std::list<ThreadWrapperPtr> worker_threads_; 
 
  std::queue<std::function<void()>> tasks_; 
  std::mutex task_mutex_; 
  std::condition_variable task_cv_; 
 
  std::atomic<int> total_function_num_; 
  std::atomic<int> waiting_thread_num_; 
  std::atomic<int> thread_id_; // 用于為新創建的線程分配ID 
 
  std::atomic<bool> is_shutdown_now_; 
  std::atomic<bool> is_shutdown_; 
  std::atomic<bool> is_available_; 
};

線程池的初始化

在構造函數中將各個成員變量都附初值，同時判斷線程池的config是否合法。

ThreadPool(ThreadPoolConfig config) : config_(config) { 
    this->total_function_num_.store(0); 
    this->waiting_thread_num_.store(0); 
 
    this->thread_id_.store(0); 
    this->is_shutdown_.store(false); 
    this->is_shutdown_now_.store(false); 
 
    if (IsValidConfig(config_)) { 
        is_available_.store(true); 
    } else { 
        is_available_.store(false); 
    } 
} 
 
bool IsValidConfig(ThreadPoolConfig config) { 
    if (config.core_threads < 1 || config.max_threads < config.core_threads || config.time_out.count() < 1) { 
        return false; 
    } 
    return true; 
}

如何開啟線程池功能?

創建核心線程數個線程，常駐于線程池，等待任務的執行，線程ID由GetNextThreadId()統一分配。

// 開啟線程池功能 
bool Start() { 
    if (!IsAvailable()) { 
        return false; 
    } 
    int core_thread_num = config_.core_threads; 
    cout << "Init thread num " << core_thread_num << endl; 
    while (core_thread_num-- > 0) { 
        AddThread(GetNextThreadId()); 
    } 
    cout << "Init thread end" << endl; 
    return true; 
}

如何關閉線程?

這里有兩個標志位，is_shutdown_now置為true表示立即關閉線程，is_shutdown置為true則表示先執行完隊列里的任務再關閉線程池。

// 關掉線程池，內部還沒有執行的任務會繼續執行 
void ShutDown() { 
    ShutDown(false); 
    cout << "shutdown" << endl; 
} 
 
// 執行關掉線程池，內部還沒有執行的任務直接取消，不會再執行 
void ShutDownNow() { 
    ShutDown(true); 
    cout << "shutdown now" << endl; 
} 
 
// private 
void ShutDown(bool is_now) { 
    if (is_available_.load()) { 
        if (is_now) { 
            this->is_shutdown_now_.store(true); 
        } else { 
            this->is_shutdown_.store(true); 
        } 
        this->task_cv_.notify_all(); 
        is_available_.store(false); 
    } 
}

如何為線程池添加線程?

這里還會檢查is_shutdown和is_shutdown_now標志，根據兩個標志位是否為true來判斷是否結束線程。

void AddThread(int id) { AddThread(id, ThreadFlag::kCore); } 
 
void AddThread(int id, ThreadFlag thread_flag) { 
    cout << "AddThread " << id << " flag " << static_cast<int>(thread_flag) << endl; 
    ThreadWrapperPtr thread_ptr = std::make_shared<ThreadWrapper>(); 
    thread_ptr->id.store(id); 
    thread_ptr->flag.store(thread_flag); 
    auto func = [this, thread_ptr]() { 
        for (;;) { 
            std::function<void()> task; 
            { 
                ThreadPoolLock lock(this->task_mutex_); 
                if (thread_ptr->state.load() == ThreadState::kStop) { 
                    break; 
                } 
                cout << "thread id " << thread_ptr->id.load() << " running start" << endl; 
                thread_ptr->state.store(ThreadState::kWaiting); 
                ++this->waiting_thread_num_; 
                bool is_timeout = false; 
                if (thread_ptr->flag.load() == ThreadFlag::kCore) { 
                    this->task_cv_.wait(lock, [this, thread_ptr] { 
                        return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() || 
                                thread_ptr->state.load() == ThreadState::kStop); 
                    }); 
                } else { 
                    this->task_cv_.wait_for(lock, this->config_.time_out, [this, thread_ptr] { 
                        return (this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() || 
                                thread_ptr->state.load() == ThreadState::kStop); 
                    }); 
                    is_timeout = !(this->is_shutdown_ || this->is_shutdown_now_ || !this->tasks_.empty() || 
                                    thread_ptr->state.load() == ThreadState::kStop); 
                } 
                --this->waiting_thread_num_; 
                cout << "thread id " << thread_ptr->id.load() << " running wait end" << endl; 
 
                if (is_timeout) { 
                    thread_ptr->state.store(ThreadState::kStop); 
                } 
 
                if (thread_ptr->state.load() == ThreadState::kStop) { 
                    cout << "thread id " << thread_ptr->id.load() << " state stop" << endl; 
                    break; 
                } 
                if (this->is_shutdown_ && this->tasks_.empty()) { 
                    cout << "thread id " << thread_ptr->id.load() << " shutdown" << endl; 
                    break; 
                } 
                if (this->is_shutdown_now_) { 
                    cout << "thread id " << thread_ptr->id.load() << " shutdown now" << endl; 
                    break; 
                } 
                thread_ptr->state.store(ThreadState::kRunning); 
                task = std::move(this->tasks_.front()); 
                this->tasks_.pop(); 
            } 
            task(); 
        } 
        cout << "thread id " << thread_ptr->id.load() << " running end" << endl; 
    }; 
    thread_ptr->ptr = std::make_shared<std::thread>(std::move(func)); 
    if (thread_ptr->ptr->joinable()) { 
        thread_ptr->ptr->detach(); 
    } 
    this->worker_threads_.emplace_back(std::move(thread_ptr)); 
}

如何將任務放入線程池中執行?

見如下代碼，將任務使用std::bind封裝成std::function放入任務隊列中，任務較多時內部還會判斷是否有空閑線程，如果沒有空閑線程，會自動創建出最多(max_threads-core_threads)個Cache線程用于執行任務。

// 放在線程池中執行函數 
template <typename F, typename... Args> 
auto Run(F &&f, Args &&... args) -> std::shared_ptr<std::future<std::result_of_t<F(Args...)>>> { 
    if (this->is_shutdown_.load() || this->is_shutdown_now_.load() || !IsAvailable()) { 
        return nullptr; 
    } 
    if (GetWaitingThreadSize() == 0 && GetTotalThreadSize() < config_.max_threads) { 
        AddThread(GetNextThreadId(), ThreadFlag::kCache); 
    } 
 
    using return_type = std::result_of_t<F(Args...)>; 
    auto task = std::make_shared<std::packaged_task<return_type()>>( 
        std::bind(std::forward<F>(f), std::forward<Args>(args)...)); 
    total_function_num_++; 
 
    std::future<return_type> res = task->get_future(); 
    { 
        ThreadPoolLock lock(this->task_mutex_); 
        this->tasks_.emplace([task]() { (*task)(); }); 
    } 
    this->task_cv_.notify_one(); 
    return std::make_shared<std::future<std::result_of_t<F(Args...)>>>(std::move(res)); 
}

如何獲取當前線程池中線程的總個數?

int GetTotalThreadSize() { return this->worker_threads_.size(); }

如何獲取當前線程池中空閑線程的個數?

waiting_thread_num值表示空閑線程的個數，該變量在線程循環內部會更新。

int GetWaitingThreadSize() { return this->waiting_thread_num_.load(); }

簡單的測試代碼

int main() { 
    cout << "hello" << endl; 
    ThreadPool pool(ThreadPool::ThreadPoolConfig{4, 5, 6, std::chrono::seconds(4)}); 
    pool.Start(); 
    std::this_thread::sleep_for(std::chrono::seconds(4)); 
    cout << "thread size " << pool.GetTotalThreadSize() << endl; 
    std::atomic<int> index; 
    index.store(0); 
    std::thread t([&]() { 
        for (int i = 0; i < 10; ++i) { 
            pool.Run([&]() { 
                cout << "function " << index.load() << endl; 
                std::this_thread::sleep_for(std::chrono::seconds(4)); 
                index++; 
            }); 
            // std::this_thread::sleep_for(std::chrono::seconds(2)); 
        } 
    }); 
    t.detach(); 
    cout << "=================" << endl; 
 
    std::this_thread::sleep_for(std::chrono::seconds(4)); 
    pool.Reset(ThreadPool::ThreadPoolConfig{4, 4, 6, std::chrono::seconds(4)}); 
    std::this_thread::sleep_for(std::chrono::seconds(4)); 
    cout << "thread size " << pool.GetTotalThreadSize() << endl; 
    cout << "waiting size " << pool.GetWaitingThreadSize() << endl; 
    cout << "---------------" << endl; 
    pool.ShutDownNow(); 
    getchar(); 
    cout << "world" << endl; 
    return 0; 
}

責任編輯：武曉燕來源：程序喵大人

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看