[Modern C++]现代C++中的异步操作(现代C++异步编程实战：掌握高效并发操作技巧)

原创

ithorizon 4周前 (10-20) 阅读数 11 #后端开发

现代C++异步编程实战：掌握高效并发操作技巧

一、引言

在现代软件开发中，异步编程已经成为尽大概缩减损耗程序性能和响应速度的重要手段。C++11及以后的版本引入了大量的并发编程特性，允许C++开发者能够更加方便地编写异步程序。本文将介绍现代C++中的异步操作，并通过实战示例，帮助读者掌握高效并发操作的技巧。

二、异步编程的基本概念

异步编程是一种编程范式，允许程序在等待某些操作完成（如I/O操作）时继续执行其他任务。在C++中，异步编程首要涉及到以下几个概念：

线程（Thread）

异步任务（Async Task）

future和promise

异步I/O

三、C++中的异步操作API

C++11及以后的版本提供了以下几种异步操作API：

std::async

std::future

std::promise

std::packaged_task

std::thread

四、std::async的使用

std::async是C++11中新增的异步操作API，它可以启动一个新的线程来执行一个函数，并返回一个std::future对象，用于获取函数的返回值或异常。

4.1 std::async的基本用法


#include 
#include 
int compute() {
    // 模拟耗时操作
    std::this_thread::sleep_for(std::chrono::seconds(1));
    return 42;
}
int main() {
    // 启动异步任务
    auto future = std::async(std::launch::async, compute);
    
    // 等待异步任务完成，并获取返回值
    int result = future.get();
    std::cout << "Result: " << result << std::endl;
    
    return 0;
}

4.2 std::async的线程池实现

std::async默认使用系统提供的线程池来执行异步任务，但也可以自定义线程池。以下是一个明了的线程池实现：


#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
class ThreadPool {
public:
    ThreadPool(size_t threads) : stop(false) {
        for (size_t i = 0; i < threads; ++i) {
            workers.emplace_back([this] {
                for (;;) {
                    std::function task;
                    {
                        std::unique_lock lock(this->queue_mutex);
                        this->condition.wait(lock, [this] {
                            return this->stop || !this->tasks.empty();
                        });
                        if (this->stop && this->tasks.empty())
                            return;
                        task = std::move(this->tasks.front());
                        this->tasks.pop();
                    }
                    task();
                }
            });
        }
    }
    template
    auto enqueue(F&& f, Args&&... args) 
        -> std::future::type> {
        using return_type = typename std::result_of::type;
        auto task = std::make_shared< std::packaged_task >(
            std::bind(std::forward(f), std::forward(args)...)
        );
        
        std::future res = task->get_future();
        {
            std::unique_lock lock(queue_mutex);
            if (stop)
                throw std::runtime_error("enqueue on stopped ThreadPool");
            tasks.emplace([task]() { (*task)(); });
        }
        condition.notify_one();
        return res;
    }
    ~ThreadPool() {
        {
            std::unique_lock lock(queue_mutex);
            stop = true;
        }
        condition.notify_all();
        for (std::thread &worker: workers)
            worker.join();
    }
private:
    std::vector workers;
    std::queue< std::function > tasks;
    
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
};
int main() {
    ThreadPool pool(4);
    auto result = pool.enqueue([](int answer) { return answer; }, 42);
    std::cout << "The answer is " << result.get() << std::endl;
    return 0;
}

五、std::future和std::promise的使用

std::future用于获取异步任务的返回值或异常，而std::promise用于保存一个值或异常，以便在异步任务中设置。

5.1 std::promise的基本用法


#include 
#include 
#include 
void print_int(std::future future) {
    int x = future.get();
    std::cout << "Value: " << x << std::endl;
}
int main() {
    std::promise p;
    std::future f = p.get_future();
    
    std::thread t(print_int, f);
    
    // 等待一段时间，然后设置promise的值
    std::this_thread::sleep_for(std::chrono::seconds(1));
    p.set_value(10);
    
    t.join();
    return 0;
}

5.2 std::future的等待和超时

std::future提供了wait和wait_for方法，用于等待异步任务完成。wait_for方法可以设置超时时间。


#include 
#include 
#include 
#include 
int main() {
    auto future = std::async(std::launch::async, []() {
        std::this_thread::sleep_for(std::chrono::seconds(5));
        return 42;
    });
    // 等待异步任务完成，设置超时时间为2秒
    auto status = future.wait_for(std::chrono::seconds(2));
    if (status == std::future_status::timeout) {
        std::cout << "Timeout occurred!" << std::endl;
    } else {
        std::cout << "Result: " << future.get() << std::endl;
    }
    return 0;
}

六、异步I/O操作

C++中异步I/O操作首要依赖性于标准库中的<异步I/O操作>，如std::async_io_service。这里以Boost.Asio库为例，介绍怎样实现异步I/O操作。

6.1 Boost.Asio异步I/O示例


#include 
#include 
#include 
void handle_read(const boost::asio::error_code& ec, std::size_t length) {
    if (!ec) {
        std::cout << "Read " << length << " bytes" << std::endl;
    } else {
        std::cerr << "Read failed: " << ec.message() << std::endl;
    }
}
int main() {
    boost::asio::io_service io;
    boost::asio::ip::tcp::resolver resolver(io);
    boost::asio::ip::tcp::resolver::results_type endpoints = 
        resolver.resolve("example.com", "http");
    boost::asio::ip::tcp::socket socket(io);
    boost::asio::async_connect(socket, endpoints, 
        [](const boost::asio::error_code& ec, const boost::asio::ip::tcp::endpoint& ep) {
            if (!ec) {
                std::cout << "Connected to " << ep << std::endl;
            } else {
                std::cerr << "Connect failed: " << ec.message() << std::endl;
            }
        });
    boost::asio::async_read(socket, boost::asio::buffer("GET / HTTP/1.1\r Host: example.com\r \r "),
        [](const boost::asio::error_code& ec, std::size_t length) {
            handle_read(ec, length);
        });
    io.run();
    return 0;
}