05-Timer.5-Synchronising completion handlers in multithreaded programs

#asio #网络编程

本教程演示了如何在多线程程序中使用 strand 类模板来同步完成处理程序。

前四个教程通过仅从一个线程调用asio::io_context::run()函数，避免了处理程序同步的问题。如你所知，asio库保证完成处理程序只会从当前正在调用asio::io_context::run()的线程中被调用。因此，仅从一个线程调用asio::io_context::run()可以确保完成处理程序不会并发执行。

在使用asio开发应用程序时，单线程方法通常是最佳起点。但其缺点是对程序（尤其是服务器）施加了限制，包括：

  • 当处理程序需要较长时间完成时，响应性较差。
  • 无法在多处理器系统上扩展性能。

如果遇到这些限制，替代方案是使用一个线程池来调用asio::io_context::run()。然而，由于这种方式允许处理程序并发执行，当处理程序可能访问共享的非线程安全资源时，我们需要采用同步机制。

#include <functional>
#include <iostream>

#include <asio.hpp>
#include <spdlog/spdlog.h>

// 我们首先定义一个名为printer的类，类似于上一个教程中的类。
// 这个类将通过并行运行两个计时器来扩展上一个教程的内容。
class Printer {
public:
    // 除了初始化一对 asio::steady_timer 成员外，该构造函数还初始化了 strand_ 成员——
    // 一个类型为 asio::strand<asio::io_context::executor_type> 的对象。
    //
    //strand 类模板是一种执行器适配器，它能确保通过它分发的处理程序在执行时，当前处理程序会先完成，才会开始下一个处理程序。
    // 无论有多少线程在调用 asio::io_context::run()，这一保证都成立。
    // 当然，这些处理程序仍可能与其他未通过 strand 分发、或通过不同 strand 对象分发的处理程序并发执行。
    Printer(asio::io_context& _io, int _n = 1)
        : m_strand(asio::make_strand(_io))
        , m_timer1(_io, std::chrono::seconds(1))
        , m_timer2(_io, std::chrono::seconds(1))
        , m_count(0)
    {
        // 在启动异步操作时，每个完成处理程序都会被"绑定"到一个asio::strand<asio::io_context::executor_type>对象上。
        // asio::bind_executor()函数会返回一个新的处理程序，该处理程序会自动通过strand对象调度其包含的处理程序。
        // 通过将处理程序绑定到同一个strand，我们可以确保它们不会并发执行。
        m_timer1.async_wait(std::bind(&Printer::print1, this, _n));
        m_timer2.async_wait(std::bind(&Printer::print2, this, _n));
    }

    ~Printer()
    {
        spdlog::info("Final count = {}", m_count);
    }

    // 在多线程程序中，若异步操作的处理程序需要访问共享资源，则应对这些处理程序进行同步。
    // 本教程中，处理程序（print1和print2）使用的共享资源是m_count数据成员。
    void print1(int _n)
    {
        if (m_count >= _n) {
            return;
        }

        m_count++;
        spdlog::info("Count = {} from print1", m_count);
        if (m_count < _n) {
            m_timer1.expires_at(m_timer1.expiry() + asio::chrono::seconds(1));
            m_timer1.async_wait(std::bind(&Printer::print1, this, _n));
        }
    }

    void print2(int _n)
    {
        if (m_count >= _n) {
            return;
        }

        m_count++;
        spdlog::info("Count = {} from print2", m_count);
        if (m_count < _n) {
            m_timer2.expires_at(m_timer2.expiry() + asio::chrono::seconds(1));
            m_timer2.async_wait(std::bind(&Printer::print2, this, _n));
        }
    }

private:
    asio::strand<asio::io_context::executor_type> m_strand;
    asio::steady_timer m_timer1;
    asio::steady_timer m_timer2;
    int m_count {};
};

// 现在的主要功能导致`asio::io_context::run()`从两个线程被调用：主线程和一个额外的线程。这是通过使用线程对象实现的。
//
// 与单线程调用时一样，只要还有“工作”要做，对`asio::io_context::run()`的并发调用就会继续执行。后台线程在所有异步操作完成之前不会退出。
int main()
{
    spdlog::info("Starting...");

    asio::io_context io;
    Printer pointer(io, 19);
    std::thread t([&io]() { io.run(); });
    io.run();
    t.join();

    return 0;
}