c++面试输出一个线程池

视频讲解输出

#pragma once

#include 
#include 
#include 
#include 
#include 
#include 
#include 

namespace yixiao
{
    

    template 
    class SpinLockQueue
    {
    public:
        void Push(T &&value)
        {
            // lock
            while (std::atomic_flag_test_and_set_explicit(&spin_lock_, std::memory_order_acquire))
                ;
            queue_.push(std::move(value));
            // unlock
            std::atomic_flag_clear_explicit(&spin_lock_, std::memory_order_release);
        }
        std::optional Pop()
        {
            // lock
            while (std::atomic_flag_test_and_set_explicit(&spin_lock_, std::memory_order_acquire))
                ;
            std::optional value;
            if (!queue_.empty())
            {
                value = queue_.front();
                queue_.pop();
            }
            // unlock
            std::atomic_flag_clear_explicit(&spin_lock_, std::memory_order_release);
            return value;
        }
        bool IsEmpty()
        {
            // lock
            while (std::atomic_flag_test_and_set_explicit(&spin_lock_, std::memory_order_acquire))
                ;
            const bool is_empty = queue_.empty();
            // unlock
            std::atomic_flag_clear_explicit(&spin_lock_, std::memory_order_release);
            return is_empty;
        }

    private:
        std::atomic_flag spin_lock_ = ATOMIC_FLAG_INIT;
        std::queue queue_;
    };

    class Semaphore
    {
    public:
        Semaphore(int n) : cnt_(n) {}
        void Down()
        {
            std::unique_lock lock(m_);
            c_.wait(lock, [&]{ return cnt_ > 0; });
            --cnt_;
        }
        void Up()
        {
            std::unique_lock lock(m_);
            ++cnt_;
            c_.notify_one();
        }

    private:
        int cnt_;
        std::mutex m_;
        std::condition_variable c_;
    };

    class ThreadPool
    {
    public:
        ThreadPool(uint64_t size) : workers_(size), is_over_(false), produce_conf_(buffer_size_), consumer_conf_(0)
        {
            for (std::thread &worker : workers_)
            {
                worker = std::thread(std::bind(&ThreadPool::WorkerThread, this));
            }
        }
        ~ThreadPool()
        {
            is_over_ = true;
            for(std::thread& worker:workers_)
            {
                if(worker.joinable())
                {
                    consumer_conf_.Up();
                    worker.join();
                }
            }
        }
        void PushTask(std::function &&f)
        {
            produce_conf_.Down();
            task_queue_.Push(std::move(f));
            consumer_conf_.Up();
        }


    private:
        void WorkerThread()
        {
            while (!is_over_.load() || !task_queue_.IsEmpty())
            {
                consumer_conf_.Down();
                const std::optional> task = task_queue_.Pop();
                if (task.has_value())
                {
                    task.value()();
                    produce_conf_.Up();
                }
                else
                {
                    consumer_conf_.Up();
                }
            }
        }
        std::atomic_bool is_over_;
        std::vector workers_;
        SpinLockQueue> task_queue_;
        Semaphore produce_conf_;
        Semaphore consumer_conf_;
        static const uint64_t buffer_size_ = 1024;
    };
}

该实现学习来自于 腾讯3TS 开源项目。