c++ 多线程编程demo

c++ 多线程编程demo

最佳实践：（摘录《c++高级编程 27.13 线程设计和最佳实践》）

1. 使用并行标准库算法
2. 终止应用程序前，确保所有thread对象都不是可joinable的
3. 最好的同步就是没有同步（各线程只读取共享数据，不写入）
4. 尝试使用单线程的所有权模式（一个线程拥有数据的独有权，处理完再传递给另一个线程）
5. 在可能时使用原子类型和操作
6. 使用锁保护可变的共享数据
7. 尽快释放锁
8. 不要手动获取多个锁，应该使用std::lock或std::try_lock(获取多个锁顺序要一致)
9. 使用多线程的分析器
10. 使用RAII锁对象（lock_guard、unique_loce、shared_lock、scoped_lock）
11. 了解调试器的对线程支持特性
12. 使用线程池，不动态创建和销毁大量线程
13. 使用高级线程库（TBB、PPL）

I/O多路复用
https://zhuanlan.zhihu.com/p/115220699

几个多线程常用demo代码：

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

using namespace std::chrono_literals;

void counter(int id, int numIterations)
{
    for (int i{ 0 }; i < numIterations; i++) {
        std::cout << "Counter " << id << " has value " << i << std::endl;
    }
}

void increment(std::atomic& counter)
{
    for (int i{ 0 }; i < 100; i++) {
        ++counter;
        std::this_thread::sleep_for(std::chrono::microseconds(1));
    }
}

void increment_ref(int& counter)
{
    std::atomic_ref atomicCounter{ counter };
    for (int i{ 0 }; i < 100; i++) {
        ++atomicCounter;
        std::this_thread::sleep_for(std::chrono::microseconds(1));
    }
}

class Counter
{
public:
    Counter(int id, int numIterations) :m_id{ id }, m_numIterations{ numIterations } {}

    void operator()() const noexcept
    {
        for (int i{ 0 }; i < m_numIterations; ++i) {
            std::lock_guard lock{ ms_mutex };
            std::cout << "Counter " << m_id << " has value " << i << std::endl;
        }
    }

    void process() {
        for (int i{ 0 }; i < m_numIterations; ++i) {
            std::unique_lock lock{ ms_timed_mutex ,200ms };
            if (lock) {
                std::cout << "Counter " << m_id << " has value " << i << std::endl;
            }
        }
    }

private:
    int m_id;
    int m_numIterations;
    inline static std::mutex ms_mutex;
    inline static std::timed_mutex ms_timed_mutex;
};

int k;
thread_local int n;

void counter_k(int id)
{
    std::cout << std::format("Thread {}:k={},n={}\n ",id, k,n);
    ++k;
    ++n;
    std::cout << std::format("Thread {}:k={},n={}\n ", id, k, n);
}

void doSomeWork()
{
    throw std::runtime_error{" Exception from thread"};
}

void threadFunc(std::exception_ptr& err)
{
    try {
        doSomeWork();
    }
    catch (...) {
        err = std::current_exception();
    }
}

void doWorkInThread()
{
    std::exception_ptr error;
    std::thread te{ threadFunc,std::ref(error) };
    te.join();
    if (error) {
        std::rethrow_exception(error);
    }
}

int main()
{
    std::cout.sync_with_stdio(false);
    std::thread t1{ counter, 1, 6 };
    std::thread t2{ counter, 2, 4 };
    t1.join();
    t2.join();

    std::thread t3{ Counter{3, 10} };
    t3.join();

    Counter c{ 4, 5 };
    std::thread t4{ std::ref(c) };
    t4.join();

    int id{ 5 };
    int numIterations{ 6 };
    std::thread t5{ [id, numIterations]() {
        for (int i{ 0 }; i < numIterations; ++i) {
            std::cout << "Counter " << id << " has value " << i << std::endl;
        }
    } };
    t5.join();

    Counter c1{ 6, 5 };
    std::thread t6{ &Counter::process,&c1 };
    t6.join();

    std::thread t11{ counter_k ,1};
    t11.join();
    std::thread t12{ counter_k ,2};
    t12.join();

    // jthread c++20 ,destory finish join
    std::jthread job{ [](std::stop_token token) {
        while (!token.stop_requested()) {
            std::cout << "running\n";
            Sleep(100);
        }
    }
    };
    Sleep(1000);
    job.request_stop();
    std::cout << "stop running\n";

    // catch error from thread
    try {
        doWorkInThread();
    }
    catch (const std::exception& e) {
        std::cout << "Main thread caught:" << e.what()<< std::endl;
    }
    // atomic
    std::atomic counter{ 0 };
    int con{0};
    std::vector threads;
    for (int i{ 0 }; i < 10; ++i) {
        threads.push_back(std::thread{ increment,std::ref(counter) });
        threads.push_back(std::thread{ increment_ref,std::ref(con) });
    }
    
    for (auto& t : threads) {
        t.join();
    }
    std::cout << "Result:" << counter << std::endl;
    std::cout << "Result:" << con << std::endl;

    std::atomic value{0};
    std::thread job1{ [&value]() {
        std::cout << "wait value:" << value << std::endl;
        value.wait(value);
        std::cout << "value:" << value << std::endl;
    } };

    std::this_thread::sleep_for(std::chrono::seconds(2));

    value = 1;
    value.notify_all();

    job1.join();

    // 同步控制 latch
    std::latch startlatch{10};
    std::vector threads_;
    for (int i{ 0 }; i < 10;i++) {
        threads_.push_back(std::jthread{ [&startlatch, id = i]() {
                std::this_thread::sleep_for(20ms);
                std::cout << "latch:" << id << std::endl;
                startlatch.count_down();

                startlatch.wait();
                std::cout << "latch:finish" << id << std::endl;
            }});
    }
    startlatch.wait();
    std::cout << "latch:finish";
    threads_.clear();

    // 同步控制 latch
    const auto workers = { "anil", "busara", "carl" };
    auto on_completion = []() noexcept {
        static const char* phase = "... done\n" \
            "Cleaning up...\n";
        std::cout << phase;
        phase = "... done\n";
    };
    std::barrier sync_point(workers.size(),on_completion);
    auto work = [&](const std::string& name) {
        std::string product = "  " + name + " worked\n";
        std::cout << product; // ok, op<< call is atomic
        sync_point.arrive_and_wait();
        product = "  " + name + " cleaned\n";
        std::cout << product;
        sync_point.arrive_and_wait();
    };

    std::cout << "Starting...\n";
    for (auto const& worker : workers) {
        threads_.push_back(std::jthread{ work,worker });
    }

    // semaphore
    threads_.clear();
    std::counting_semaphore sp{4};
    for (int i{ 0 }; i < 20; i++) {
        threads_.push_back(std::jthread{ [&sp,id = i]() {
            sp.acquire();
            std::cout << "thread..."<< id << "\n";
            std::this_thread::sleep_for(100ms);
            sp.release();
            }
            });
    }

    // future and promise
    std::promise m_;
    std::future theFuture{m_.get_future()};
    std::thread t{ [](std::promise p) { p.set_value(12); }, std::move(m_) };
    auto va = theFuture.get();
    std::cout << "the future value:" << va << std::endl;
    t.join();
  
    // 条件变量
    std::mutex m_mu;
    std::unique_lock lock(m_mu);
    std::condition_variable con_var;
    std::atomic ca{0};
    std::async(std::launch::async, [&ca]() {
        std::cout << "thread the value in: \n";
        std::this_thread::sleep_for(3s);
        std::cout << "change the value end: \n";
        ca = 5;
        });
    std::cout << "wait the value: \n";
    con_var.wait(lock, [&ca]() { return (ca > 0); });
    std::cout << "wait the value finish:" << ca << std::endl;

    // packaged_task
    std::packaged_task task{ []() { return 12; } };
    std::future fur{ task.get_future() };
    std::thread tt{std::move(task)};
    auto re = fur.get();
    std::cout << "wait the task value:" << re << std::endl;
    tt.join();

    // shared_future
    std::promise tP1, tP2;
    std::promise signalP;
    std::shared_future sf{ signalP.get_future().share() };
    auto fun1 = [&tP1, sf]() {
        tP1.set_value();
        auto p{sf.get()};
        std::cout << "thread1 get value:" << p << std::endl;
    };
    auto fun2 = [&tP2, sf]() {
        tP2.set_value();
        auto p{ sf.get() };
        std::cout << "thread1 get value:" << p << std::endl;
    };
     
    auto res2{ std::async(std::launch::async,fun2) };

    tP1.get_future().wait();
    tP2.get_future().wait();

    signalP.set_value(22);

    return 1;
}
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289