C++11 异步任务实现（类Java ExecutorService）

发表于 分类于

使用Java时，并发库提供很多有用的工具，比如ExecutorService，方便执行异步任务。C++11通过使用线程和lambda，可以很容易实现类似的功能。尽在代码中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class AsyncService {
    MZ_DISALLOW_COPY_AND_ASSIGN(AsyncService)

    using Async_t = std::function<void()>;
    using AsyncQueue_t = std::queue<Async_t>;

private:
    AsyncThread** m_asyncThreads = nullptr;
    int m_threadSize = 0;

    std::mutex m_asyncMutex;
    std::condition_variable m_asyncCond;
    AsyncQueue_t m_asyncQueue;

public:
    AsyncService();

    ~AsyncService();

    bool Initialize(int threadSize);

    void Finalize();

    void Push(Async_t async);

    Async_t Pop();
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
AsyncService::AsyncService() {}

AsyncService::~AsyncService() {}

bool AsyncService::Initialize(int threadSize) {
    m_threadSize = threadSize;
    m_asyncThreads = new AsyncThread*[m_threadSize];
    for (auto index = 0; index < m_threadSize; ++index) {
        m_asyncThreads[index] = new AsyncThread(*this);
        m_asyncThreads[index]->Initialize();
    }

    return true;
}

void AsyncService::Finalize() {
    for (auto index = 0; index < m_threadSize; ++index) {
        m_asyncThreads[index]->Finalize();
        SafeDelete(m_asyncThreads[index]);
    }
    SafeDeleteArray(m_asyncThreads);
    m_threadSize = 0;
}

void AsyncService::Push(Async_t async) {
    std::unique_lock<std::mutex> lock(m_asyncMutex);
    m_asyncQueue.push(async);
    m_asyncCond.notify_all();
}

AsyncService::Async_t AsyncService::Pop() {
    std::unique_lock<std::mutex> lock(m_asyncMutex);
    m_asyncCond.wait(lock, [this] {
        return !m_asyncQueue.empty();
    });

    auto async = m_asyncQueue.front();
    m_asyncQueue.pop();
    return async;
}

使用方式很简单，调用Push接口即可。

1
2
3
asyncService.Push([&] {
    // Do something
});

AsyncService 提供异步任务功能接口，使用 mutex 和 condition_variable 实现多线程任务队列同步。线程依次从队列中取出任务并执行，为方便使用，任务使用lambda函数。