Asio17-MultiThreadPool

Asio17-MultiThreadPool.md

区别

Asio-16也是一种多线程模式,那么他跟这一节有什么区别呢?

简单而言,上一节的结构是,n个线程,每个线程一个io_context在运行,相当于底层多个epoll.这一节呢,也是n个线程,但是只有1个线程有io_context在run,底层相当于一个epoll.

区别在于我们上一节的多线程,每一个线程在各自的io_context取出回调,交给上层的逻辑层去处理,每个线程只管自己的一部分,没有线程安全问题.

这一节的多线程呢,每一个线程都在一个io_context等待着取出回调,然后交给上层的逻辑层.但是多个线程同时在一个任务队列取回调,必然要引发线程安全问题的.那么怎么做呢?我们使用了asio提供的boost::asio::strand来实现串行的取出回调.

结构如下:

首先实现ThreadPool:

// .h
#include "Singleton.hpp"#include <boost/asio.hpp>class ThreadPool : public Singleton<ThreadPool> {
public:friend class Singleton<ThreadPool>;~ThreadPool();boost::asio::io_context& GetContext();void Stop();private:ThreadPool(int num = std::thread::hardware_concurrency());boost::asio::io_context _io_context;std::unique_ptr<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> _work_guard;std::vector<std::thread> _threads;
};// .cpp
#include "ThreadPool.h"
#include <boost/asio/io_context.hpp>void ThreadPool::Stop()
{_work_guard.reset();_io_context.stop();for (auto& thread : _threads) {thread.join();}
}ThreadPool::ThreadPool(int num): _work_guard(new boost::asio::executor_work_guard<boost::asio::io_context::executor_type>(_io_context.get_executor()))
{for (std::size_t i = 0; i < num; ++i) {_threads.emplace_back([this]() {_io_context.run();});}
}boost::asio::io_context& ThreadPool::GetContext()
{return _io_context;
}ThreadPool::~ThreadPool()
{Stop();
}

同时CSession添加一个变量:

strand<io_context::executor_type> _strand;

其实流程还是有点难以理解,到底怎么多线程的?

我个人理解,每个CSession都会有一个独立的变量strand,而这个stand类似一个队列,我们看到异步读写的回调函数是这样调用的:

boost::asio::async_read(_socket, boost::asio::buffer(_recv_msg_node->_data, data_len), boost::asio::bind_executor(_strand, std::bind(&CSession::HandleMsg, this, std::placeholders::_1, std::placeholders::_2, SharedSelf())));

实际上当有回调函数的时候,这个任务就会进入这个strand队列之中,而由于所有的CSession里面的strand初始化的时候都绑定了同一个io_context,而这个io_context跑在多个线程之中,那么asio就会分配线程去处理strand中的回调.

实际类似这样:

// 创建3个session
Session session1(io), session2(io), session3(io);// 提交任务
session1.操作("A1");  // strand1队列: [A1]
session1.操作("A2");  // strand1队列: [A1] → [A2]session2.操作("B1");  // strand2队列: [B1]  
session2.操作("B2");  // strand2队列: [B1] → [B2]session3.操作("C1");  // strand3队列: [C1]
session3.操作("C2");  // strand3队列: [C1] → [C2]// 4个线程执行io_context.run()，可能的执行情况：
线程1: 取出strand1的A1执行 → 完成后取出strand1的A2执行
线程2: 取出strand2的B1执行 → 完成后取出strand2的B2执行  // 与线程1并行！
线程3: 取出strand3的C1执行 → 完成后取出strand3的C2执行  // 与其他并行！
线程4: 空闲或执行其他任务

也就是说,每个线程之内是串行的完成队列里的各个任务,而线程间是并行的!

根据测试呢,发现,ThreadServices版本比ThreadPool版本要快1秒左右(前者4s,后者5-6s,客户端100个线程,每个线程500个循环,收发包).

客户端代码

#include <boost/asio.hpp>
#include <chrono>
#include <iostream>
#include <json/json.h>
#include <json/reader.h>
#include <json/value.h>
#include <thread>
#include <vector>using namespace std;
using boost::asio::ip::tcp;constexpr int MAX_LENGTH = 1024 * 2;
constexpr int HEAD_TOTAL = 4; // 2字节id + 2字节len/*---------- 主函数 ----------*/
int main()
{auto start = chrono::high_resolution_clock::now();vector<thread> thr;for (int i = 0; i < 100; ++i) {thr.emplace_back([i] {try {boost::asio::io_context ioc;tcp::socket sock(ioc);sock.connect(tcp::endpoint { {}, 9999 });for (int seq = 0; seq < 500; ++seq) {/* 1. 构造 JSON 正文 */Json::Value root;root["id"] = 1001;root["msg"] = "hello world";string body = root.toStyledString();/* 2. 打包头部 + 正文 */uint16_t msg_id = 1001;uint16_t body_len = static_cast<uint16_t>(body.size());unsigned char out[HEAD_TOTAL + body.size()];memcpy(reinterpret_cast<char*>(out), &msg_id, 2);memcpy(reinterpret_cast<char*>(out) + 2, &body_len, 2);memcpy(reinterpret_cast<char*>(out) + 4, body.data(), body.size());/* 3. 一次性写出 */boost::asio::write(sock, boost::asio::buffer(reinterpret_cast<char*>(out), HEAD_TOTAL + body.size()));/* 4. 收头部 */unsigned char head[HEAD_TOTAL];boost::asio::read(sock, boost::asio::buffer(head, HEAD_TOTAL));uint16_t reply_id = 0, reply_len = 0;memcpy(&reply_id, head, 2);memcpy(&reply_len, head + 2, 2);if (reply_len == 0 || reply_len > MAX_LENGTH - 1) {cerr << "bad reply_len=" << reply_len << '\n';break;}/* 5. 收正文 */// vector<unsigned char> payload(reply_len);char payload[MAX_LENGTH] = { 0 };boost::asio::read(sock, boost::asio::buffer(payload, reply_len));/* 6. 解析 JSON */Json::Value reply;string payload_str(payload, reply_len);if (!Json::Reader {}.parse(payload_str, reply)) {cerr << "json parse fail\n";continue;}cout << "id=" << reply["id"]<< " msg=" << reply["msg"].asString()<< " thr=" << i<< " seq=" << seq << '\n';}} catch (exception& e) {cerr << "Exception: " << e.what() << '\n';}});this_thread::sleep_for(chrono::milliseconds(10));}for (auto& t : thr)t.join();auto dur = chrono::duration_cast<chrono::seconds>(chrono::high_resolution_clock::now() - start);cout << "Time spent: " << dur.count() << " s\n";return 0;
}

这里其实很不"现代",用了很多裸指针,裸数组.主要是为了兼容之前的c风格.如果cpp风格的话,可以使用vector完全替代裸数组.