目录

boost::asio::io_context 是 Boost.Asio 库中用于处理异步操作的核心类。它管理 I/O 任务的调度和执行，提供了一种机制，允许程序员在不阻塞主线程的情况下执行异步操作。

其主要作用如下：

  1. 任务调度

    io_context 提供了一个事件循环机制。通过它，可以调度多个 I/O 操作，等待它们完成，并在操作完成时执行回调函数。这使得程序可以高效地处理大量异步 I/O 请求，而不必阻塞主线程。

  2. 执行异步操作

    它是异步操作的入口。无论是定时器、套接字读写还是其他异步操作，通常都需要将这些操作与 io_context 关联，并通过它来执行。

  3. 管理工作线程

    当多个线程需要处理 I/O 请求时，io_context 可以在线程间安全共享，并通过 run() 方法在不同的线程中执行任务。这为多线程异步编程提供了良好的支持。

  4. 事件循环

    io_context::run() 方法会启动事件循环，监听所有的 I/O 事件，并在有任务时执行相关回调函数。当没有任务时，run() 方法会阻塞，直到新的任务被调度。

主要方法

  • io_context::run()：启动事件循环，执行任务队列中的任务，直到任务全部完成。
  • io_context::stop()：停止事件循环，任何等待中的任务将不再被执行。
  • io_context::post()：将一个函数投递到 io_context，等待事件循环执行。

代码示例

#include <iostream>
#include <boost/asio.hpp>

void print_message() {
    std::cout << "Hello from io_context!" << std::endl;
}

int main() {
    boost::asio::io_context io;

    // 使用 post 将任务添加到 io_context
    io.post(&print_message);

    // 启动事件循环，执行任务
    io.run();

    return 0;
}

在这个示例中，我们使用 io_context::post() 将 print_message 函数添加到 io_context 的任务队列中，然后通过 io_context::run() 启动事件循环来执行任务。

总结

boost::asio::io_context 是 Boost.Asio 库中的核心组件，用于管理异步 I/O 操作。通过事件循环机制，它可以调度并执行异步任务，提供高效的 I/O 处理能力。

在rest和grpc的server中使用(不建议)

flowchart TD A[Start main] --> B[Initialize glog] B --> C[Create DAO and Service] C --> D[Create boost::asio::io_context] D --> E[Start gRPC Thread] D --> F[Start HTTP Thread] E --> G[run_grpc_server] F --> H[run_http_server] G --> G1[Configure gRPC Server] G1 --> G2[Create GrpcHandler] G2 --> G3[Add Listening Port] G3 --> G4[Register Service] G4 --> G5[Build and Start Server] G5 --> G6[Create work_guard] G6 --> G7[io_context.run] H --> H1[Create HTTP Listener] H1 --> H2[Create HttpHandler] H2 --> H3[Setup Routes] H3 --> H4[Open Listener] H4 --> H5[Create work_guard] H5 --> H6[io_context.run] G7 --> I[Handle gRPC Requests] H6 --> J[Handle HTTP Requests] E --> K[grpc_thread.join] F --> L[http_thread.join] K & L --> M[Shutdown]

为什么 io_context.run() 可以奏效？

io_context.run() 的作用是启动一个事件循环，它会持续运行，直到所有与 io_context 关联的任务被处理完毕。在你的代码中，无论是 gRPC 服务器还是 HTTP 服务器，都依赖 io_context 来管理它们的 I/O 操作，这使得 io_context 负责协调和调度这些任务。

  • gRPC 和 HTTP 服务器的事件处理：在 run_grpc_server 和 run_http_server 函数中，boost::asio::executor_work_guard 确保 io_context 不会在没有任务时立即退出。即使当前没有任何任务，run() 方法会保持事件循环的运行，等待新任务（如 gRPC 和 HTTP 请求）到来。
  • 异步任务调度：当有请求（如网络连接、数据读取或写入）到来时，这些请求将被注册到 io_context 中，io_context.run() 会处理这些请求并调用相应的回调函数。因为 gRPC 和 HTTP 的 I/O 操作是异步的，io_context 可以有效地管理这些异步操作。

请求处理是在同一个线程还是线程池？

在你的代码中，io_context.run() 是在多个线程中执行的，具体如下：

  1. gRPC 服务器和 HTTP 服务器在不同线程中执行：

    你创建了两个线程：一个用于 gRPC 服务器（grpc_thread），另一个用于 HTTP 服务器（http_thread）。每个线程分别调用 io_context.run()，这意味着这两个线程会并行运行，并处理各自的任务。

  2. 同一个 io_context 共享：

    两个服务器都共享同一个 boost::asio::io_context 实例，因此无论是 gRPC 还是 HTTP 请求，它们的 I/O 操作都会在这个 io_context 中调度。因为 io_context 是线程安全的，它能够在不同线程中被并发调用。

  3. 多个线程处理请求：

    当有新请求进来时，这些请求会被调度到 io_context 中，然后由调用 run() 的线程（在本例中是 grpc_thread 和 http_thread）进行处理。因此，请求会在不同的线程中被处理，而不是都集中在一个线程中。这相当于一个简易的线程池机制，两个线程同时从 io_context 获取并处理任务。

代码运行的关键流程

  • boost::thread 创建的线程分别运行 gRPC 和 HTTP 服务器的事件循环。
  • 这两个线程都共享同一个 io_context，通过 io_context.run() 处理 I/O 事件。
  • 当有新请求（如 HTTP 或 gRPC 请求）到达时，这些请求被注册到 io_context 中，由调用 run() 的线程（grpc_thread 和 http_thread）处理。
  • 如果有更多线程加入 io_context 的事件循环，则可以提高并发处理能力。

总结

  • io_context.run() 启动了事件循环，负责调度和执行所有注册的异步任务。
  • 请求在多个线程中并行处理，每个线程都会调用 io_context.run() 来处理来自 io_context 的任务。

实际上不推荐grpc的线程和http的线程用同一个io_context，因为grpc内部本身也有线程池。

#include <iostream>
#include <memory>
#include <thread>
#include <grpcpp/grpcpp.h>
#include <glog/logging.h>
#include <boost/asio.hpp>
#include <boost/thread/thread.hpp>
#include "interface/grpc_handler.h"
#include "interface/http_handler.h"
#include "service/hello_service.h"
#include "dao/hello_dao.h"

// gRPC服务器运行函数
void run_grpc_server(std::shared_ptr<HelloService> service, boost::asio::io_context& io_context) {
    // 设置gRPC服务器地址和端口
    std::string server_address("0.0.0.0:50051");
    // 创建gRPC处理器实例
    GrpcHandler handler(service);

    // 配置gRPC服务器
    grpc::ServerBuilder builder;
    // 添加监听端口，使用不安全的证书（仅用于测试）
    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
    // 注册服务
    builder.RegisterService(&handler);

    // 构建并启动服务器
    std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
    LOG(INFO) << "gRPC server listening on " << server_address;
    // 使用boost::asio::io_context来运行服务器
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work_guard(io_context.get_executor());
    io_context.run();
}

// HTTP服务器运行函数
void run_http_server(std::shared_ptr<HelloService> service, boost::asio::io_context& io_context) {
    // 创建HTTP监听器
    web::http::experimental::listener::http_listener listener("http://0.0.0.0:8080");
    // 创建HTTP处理器实例
    HttpHandler handler(service);
    // 设置路由
    handler.setup_routes(listener);

    try {
        // 打开监听器并等待
        listener.open().wait();
        LOG(INFO) << "HTTP server listening on http://0.0.0.0:8080";
        
        // 使用boost::asio::io_context来保持服务器运行
        boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work_guard(io_context.get_executor());
        io_context.run();
    }
    catch (const std::exception & e) {
        // 捕获并打印任何异常
        LOG(ERROR) << "Exception in HTTP server: " << e.what();
    }
}

int main(int argc, char* argv[]) {
    // 初始化 glog
    google::InitGoogleLogging(argv[0]);
    // 将日志输出到 stderr
    FLAGS_logtostderr = 1;

    LOG(INFO) << "Starting server...";

    // 创建DAO层实例
    auto dao = std::make_shared<HelloDao>();
    // 创建服务层实例，注入DAO依赖
    auto service = std::make_shared<HelloService>(dao);

    // 创建boost::asio::io_context
    boost::asio::io_context io_context;

    // 创建并启动gRPC服务器线程
    boost::thread grpc_thread(boost::bind(run_grpc_server, service, boost::ref(io_context)));
    // 创建并启动HTTP服务器线程
    boost::thread http_thread(boost::bind(run_http_server, service, boost::ref(io_context)));

    LOG(INFO) << "Server threads started";

    // 等待两个线程结束
    grpc_thread.join();
    http_thread.join();

    LOG(INFO) << "Server shutting down";

    return 0;
}

rest 线程池以及通过io_context.run()分发

#include <iostream>
#include <memory>
#include <vector>
#include <thread>
#include <glog/logging.h>
#include <gflags/gflags.h>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/filesystem.hpp>  // 使用 Boost 的 filesystem 库
#include "interface/http_handler.h"
#include "service/hello_service.h"
#include "dao/hello_dao.h"
#include "common.h"

// 定义命令行参数
DEFINE_int32(num_threads, 32, "Number of threads for the thread pool");
DEFINE_string(http_address, "http://0.0.0.0:8080", "HTTP server address");



// HTTP服务器运行函数
void run_http_server(std::shared_ptr<HelloService> service, boost::asio::io_context& io_context) {
    // 创建HTTP监听器
    web::http::experimental::listener::http_listener listener(FLAGS_http_address);
    // 创建HTTP处理器实例
    HttpHandler handler(service);
    // 设置路由
    handler.setup_routes(listener);

    try {
        // 打开监听器并等待
        listener.open().wait();
        LOG_INFO("HTTP server listening on " << FLAGS_http_address);
        
        // 创建 work_guard 以保持 io_context 运行
        auto work_guard = boost::asio::make_work_guard(io_context);
        
        // 运行 io_context
        io_context.run();
    }
    catch (const std::exception & e) {
        LOG_ERROR("Exception in HTTP server: " << e.what());
    }
    LOG_INFO("Http Server shutting down");
}

int main(int argc, char* argv[]) {
    // 初始化 gflags
    gflags::ParseCommandLineFlags(&argc, &argv, true);
    
    // 初始化 glog
    google::InitGoogleLogging(argv[0]);
    
    // 设置日志目录
    if (!FLAGS_log_dir.empty()) {
        // 确保日志目录存在
        boost::filesystem::create_directories(FLAGS_log_dir);
        
        // 设置日志文件名前缀
        google::SetLogDestination(google::INFO, (FLAGS_log_dir + "/info_").c_str());
        google::SetLogDestination(google::WARNING, (FLAGS_log_dir + "/warning_").c_str());
        google::SetLogDestination(google::ERROR, (FLAGS_log_dir + "/error_").c_str());
        
        // 设置 vlog 文件名前缀
        google::SetVLOGLevel("*", 0);  // 设置所有文件的 VLOG 级别为 0
        google::SetStderrLogging(google::INFO);  // 设置标准错误输出的最低日志级别
    }
    
    // 同时输出到文件和 stderr
    FLAGS_alsologtostderr = true;
    
    // 设置日志前缀格式，包含线程ID
    FLAGS_log_prefix = true;

    LOG_INFO("Starting server...");

    // 创建DAO层实例
    auto dao = std::make_shared<HelloDao>();
    // 创建服务层实例，注入DAO依赖
    auto service = std::make_shared<HelloService>(dao);

    // 创建boost::asio::io_context
    boost::asio::io_context io_context;

    // 创建 work_guard 以保持 io_context 运行
    auto work_guard = boost::asio::make_work_guard(io_context);

    // 创建并启动HTTP处理线程池
    std::vector<std::thread> http_handler_threads;
    for (int i = 0; i < FLAGS_num_threads; ++i) {
        http_handler_threads.emplace_back([&io_context]() { 
            io_context.run(); 
        });
    }

    // 创建并启动HTTP服务器设置线程
    std::thread http_setup_thread(run_http_server, service, std::ref(io_context));

    LOG_INFO("Server threads started");

    // 等待线程结束（实际上它们不会结束，除非发生错误或手动终止）
    http_setup_thread.join();
    for (auto& t : http_handler_threads) {
        t.join();
    }

    LOG_INFO("Server shutting down");

    return 0;
}

在这段代码中，HTTP 请求的处理方式是通过 Boost.Asio 的 io_context 进行异步处理，并利用多线程模型来并发执行任务。你提到的路由分发机制如下：

1. HTTP 路由的设置 (setup_routes)

在 HTTP 服务器的启动过程中，首先会通过 handler.setup_routes(listener) 来设置路由。路由是 URL 与相应的处理逻辑之间的映射。例如，如果收到一个 /hello 的 HTTP 请求，系统可以将其映射到某个处理函数。

handler.setup_routes(listener);

这个步骤是由 run_http_server 中的 http_setup_thread 启动的。可以理解为，它设置了一个基本的路由规则，用于将不同的 HTTP 请求分发到对应的处理器函数。在这一阶段，服务器还没有真正开始处理请求。

2. io_context** 与线程池**​

io_context 才是核心的 I/O 调度机制，它负责管理所有异步 I/O 操作。线程池通过多个线程运行 io_context.run()，这些线程将处理实际的 HTTP 请求。

for (int i = 0; i < FLAGS_num_threads; ++i) {
    http_handler_threads.emplace_back([&io_context]() { 
        io_context.run(); 
    });
}

io_context 中包含了所有注册的异步操作（如 HTTP 请求的接收和处理）。这些异步操作通过线程池分发给不同的线程处理。

3. HTTP 请求的处理流

  • 第一步：路由设置：当 HTTP 服务器启动时，handler.setup_routes(listener) 会配置路由表，决定不同的 URL 请求应该由哪个处理器处理。
  • 第二步：请求进入io_context：当客户端发送 HTTP 请求时，boost::asio::io_context 将该请求注册为一个异步任务，并放入任务队列中等待处理。
  • 第三步：线程处理请求：线程池中的多个线程通过 io_context.run() 从任务队列中取出任务进行处理。每个线程处理一个请求，调用相应的处理器（根据之前设置的路由）来完成业务逻辑。

4. 路由与线程池的协作关系

  • 路由设置线程 (setup_routes)：这是在启动时由 http_setup_thread 完成的，它负责配置路由表。这个过程是同步的，只会执行一次。
  • 请求处理线程 (****io_context** 线程池)**：当路由设置完成后，所有的 HTTP 请求会交给 io_context，然后通过多线程并发处理。这意味着请求首先通过 setup_routes 配置的路由分发到对应的处理函数，然后实际的请求处理是由多个线程执行的。

总结

  • 路由 (setup_routes) 只是配置阶段：路由设置发生在 HTTP 服务器启动时，它并不参与实际的请求处理过程，只是决定请求的 URL 应该对应哪个处理函数。
  • 线程池 (****io_context** 线程池) 处理实际请求**：当路由设置完成后，所有的 HTTP 请求会被 io_context 分发到线程池中，并由多个线程并发执行请求处理。

因此，路由配置线程 (setup_routes) 和请求处理线程 (****io_context** 线程池)** 是分开的，路由只在启动时起作用，而实际请求的处理是并发执行的。

线程角色总结：

  • http_setup_thread：
    • 负责启动 HTTP 服务器和配置 io_context。
    • 在 io_context.run() 被调用后，它可能会被阻塞，等待服务器关闭或出错。
    • 它在正常情况下，保持 io_context 活动，但实际上不处理任何 HTTP 请求。
  • http_handler_threads：
    • 这些线程是实际处理 HTTP 请求的工作线程。
    • 它们从 io_context 的任务队列中获取任务（HTTP 请求），处理这些请求并生成响应。
    • 线程池中的线程通过调用 io_context.run() 处理由 io_context 分配的任务。
graph TD A[主线程] --> B[启动 http_setup_thread] B --> C[启动 HTTP 服务器] C --> D[调用 io_context.run] D --> E[等待请求] F[http_handler_threads] --> G[处理 HTTP 请求] E --> G classDef setup fill:#f9f,stroke:#333,stroke-width:2px; class B,C setup;
graph TD A[启动 HTTP 服务器] --> B[调用 setup_routes 配置路由] B --> C[启动 io_context 和线程池] C --> D[客户端发送 HTTP 请求] D --> E{请求匹配路由?} E -->|是| F[分配到相应的处理器] E -->|否| G[返回 404 Not Found] F --> H[请求加入 io_context 队列] H --> I[线程池中的线程从队列中获取任务] I --> J[处理请求并返回响应]