C++11异步任务轮子实现（header-only）

为什么写这个

1. C++17异步任务需要future和promise配合使用，不是很喜欢那种语法。
2. 实现一个操作简洁的异步任务。

满足功能

1. 异步任务
2. 超时控制
3. get接口同步
4. 任务计时
5. lambda回调
6. 任务重启

使用

#include "async_callback.h"
#include 
#include 
using namespace std;

int main() {
    int a = 0, b = 0, c = 0;
    AsyncTask* task1 = new AsyncTask([&a](){
        sleep(2);
        for (int i = 0; i < 10000; ++i) {
            ++a;
        }
    });
    AsyncTask* task2 = new AsyncTask([&b](){
        sleep(2);
        for (int i = 0; i < 10000; ++i) {
            ++b;
        }
    });

    if (!task1->get(1)) {
        cout << "task1超时" << endl;
    }
    if (task2->get()) {
        cout << "task2没超时" << endl;
    }
    task1->restart();
    task2->restart();
    task1->get();
    task2->get();
    cout << a << endl;
    cout << b << endl;
    cout << "task1执行时间:" << task1->executionTime() << endl;
    cout << "task2执行时间:" << task2->executionTime() << endl;
    return 0;
}
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

打印结果：

task1超时
task2没超时
20000
20000
task1执行时间:2.00009
task2执行时间:2.00009
1
2
3
4
5
6

代码

// Simple asynchronous tasks with timeout
// Author: Y. F. Zhang
// Date: 2023-09-21

#ifndef ASYNC_CALLBACK_H
#define ASYNC_CALLBACK_H

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
class Timer {
public:
    Timer() {
        offsetTime_ = 0;
        timerState_ = TIMERSTOP;
    }
    void timerStart() {
        timerState_ = TIMERRUN;
        timeStamp_ = getWallTime();
    }
    double elapsedTime() {
        if (timerState_ == TIMERPAUSE) {
            return offsetTime_;
        }
        curTime_ = getWallTime() - timeStamp_ + offsetTime_;
        if (timerState_ == TIMERSTOP) return 0;
        return curTime_;
    }
    void pauseTimer() {
        offsetTime_ += getWallTime() - timeStamp_; 
        timerState_ = TIMERPAUSE;
    }
    void stopTimer() {
        offsetTime_ = 0;
        curTime_ = 0;
        timerState_ = TIMERSTOP;
    }


private:
    double getWallTime() {
        struct timeval time ;
        if (gettimeofday(&time,NULL)){
            return 0;
        }
        return (double)time.tv_sec + (double)time.tv_usec * .000001;
    }
    double timeStamp_;
    double curTime_;
    double offsetTime_;

    enum{
        TIMERSTOP,
        TIMERRUN,
        TIMERPAUSE
    } timerState_;
};

namespace AsyncTaskException {

class GetButTaskStopErr: public std::exception {
    const char* what() const throw () {
        return "invoke get method but task is stop!";
    }
};

class GetExecutionTimeButTaskStopErr: public std::exception {
    const char* what() const throw () {
        return "invoke executionTime method but task is stop!";
    }
};
 
}

using func = std::function<void()>;
class AsyncTask {
private:
    enum {
        TASKRUNNING,
        TASKSTOP,
        TASKFINISHED
    } taskState_;
    func callback_;
    std::mutex mtx_;
    std::condition_variable cond_;
    std::atomic_bool completedFlag_;
    std::thread* taskThread_;
    Timer timer_;
    void initTask() {
        timer_.stopTimer();
        taskState_ = TASKSTOP;
        completedFlag_.store(false);
        if (taskThread_ != nullptr) {
            delete taskThread_;
            taskThread_ = nullptr;
        }
    }
    void runTask() {
        auto wrapperCallback = [this]() {
            try {
                timer_.timerStart();
                callback_();
                timer_.pauseTimer();
                completedFlag_.store(true);
            } catch (std::exception e) {
                fprintf(stderr, "%s", e.what());
            }
            taskState_ = TASKFINISHED;
            cond_.notify_one();
        };
        taskState_ = TASKRUNNING;
        taskThread_ = new std::thread(wrapperCallback);
        taskThread_->detach();
    }
public:
    AsyncTask(func&& callback) {
        this->callback_ = callback;
        restart();
    }
    void restart() {
        initTask();
        runTask();
    }
    double executionTime() {
        if (taskState_ == TASKSTOP) {
            throw AsyncTaskException::GetExecutionTimeButTaskStopErr();
        }
        return timer_.elapsedTime();
    }
    ~AsyncTask() {
        if (taskThread_ != nullptr) {
            delete taskThread_;
        }
    }
    bool get(size_t timeoutSec = 0) {
        if (taskState_ == TASKSTOP) {
            throw AsyncTaskException::GetButTaskStopErr();
        }
        std::unique_lock<std::mutex> lk(mtx_);
        if (timeoutSec == 0) {
            cond_.wait(lk, [this](){return completedFlag_.load();});
        } else {
            return cond_.wait_for(lk, std::chrono::seconds(timeoutSec), [this](){return completedFlag_.load();});
        }
        return true;
    }

};

#endif
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156