dl_thread_task.h

浏览该文件的文档.

#pragma once
 
#include <mutex>
 
#include "system/dl_thread.h"
#include "base/dl_list_mt_cache.h"
 
namespace dl
{

template<typename T>
class MultiThreadTask
{
public:
 
    using FnDoTask = std::function<void(T&)>;
    
    void Init(FnDoTask fn_do_task, size_t thread_num = (size_t)std::thread::hardware_concurrency())
    {
        _list.Clear();
 
        _fnDoTask = fn_do_task;
        _bEnd = false;
 
        _allThread.resize(thread_num);
        for (auto& iter : _allThread)
        {
            iter = g_thread->Create();
            iter->Start(std::bind(&MultiThreadTask::_thread_task, this), nullptr);
        }
    }

    void AddTask(const T& task)
    {
        ++_numTask;
        _list.Add(task);
    }

    void ClearTask()
    {
        // todo
        /*_list.Clear();
        _numDoTask = 0;*/
    }

    size_t GetTaskNum()
    {
        return _numTask;
    }

    void Wait()
    {
        _list.Move();
        while (!_bEnd)
        {
            if (_list.IsEmpty() && _numTask == 0)
                break;
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
    }
 
    void SetQueueMode(QueueMode mode)
    {
        _list.SetQueueMode(mode);
    }
 
    MultiThreadTask() :
        _numTask{ 0 }
    {
 
    }
 
    ~MultiThreadTask()
    {
        _bEnd = true;
        for (auto& iter : _allThread)
        {
            delete iter;
        }
    }
private:
    ListMultiThreadCache<T> _list;
    FnDoTask _fnDoTask;
    std::atomic_int _numTask; // 剩余任务数
    std::vector<Thread*> _allThread;
    std::atomic_bool _bEnd;
 
    void _thread_task()
    {
        while (!_bEnd)
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
 
            T task;
            if (_list.PopFront(task))
            {
                _fnDoTask(task);
                --_numTask;
            }
        }
    }
};
}