基于libevent的异步事件驱动型线程池实现

                             +----------------------+|      IFoxThread      |  ← 抽象线程接口|----------------------|| + dispatch()         || + start()            || + stop()             || + ...                |+----------^-----------+|--------------------|----------------------|                                         |+------------------------+              +---------------------------+|       FoxThread        |              |      FoxThreadPool        ||------------------------|              |---------------------------|| 单个线程 + event_loop  |               | 多线程池 + 回调分发机制     || 内部维护 thread/event  |               | 维护线程列表及任务队列      || 支持任务 dispatch     |                | 支持负载均衡分发任务        |+------------------------+              +---------------------------+^                                         ^|                                         ||               +-------------------------+|               |+-----------------------------------+|        FoxThreadManager           ||-----------------------------------||  线程调度管理器，支持按名称查找    ||  维护 map<string, IFoxThread::ptr>||  可分发/广播任务至指定线程或池中   ||  start()/stop() 控制所有子线程     |+-----------------------------------+

threadPool.h

/*** @file threadPool.h* @brief 实现和管理线程池和线程调度* @date 2025-05-11* @copyright Copyright (c) 2025 All rights reserved.*/#ifndef __MYPROJECT_THREAD_POOL_H
#define __MYPROJECT_THREAD_POOL_H#include <thread>
#include <vector>
#include <string>
#include <list>
#include <map>
#include <functional>
#include <memory>
#include <event2/bufferevent.h>
#include <event2/buffer.h>
#include <event2/listener.h>#include "myproject/singleton.h"
#include "myproject/mutex.h"namespace myproject {class FoxThread;
// 这个 IFoxThread 类是一个线程调度接口类，属于 myproject 框架中的事件线程调度模块，
// 它为不同的线程模型（如单线程、线程池）提供统一的调度方式。IFoxThread 是一个
// 纯虚类，定义了一组线程调度的接口，供其派生类（如 FoxThread、FoxThreadPool）实现。
// IFoxThread 用于抽象线程调度操作，统一事件派发、线程控制和状态输出的接口，
// 让用户无需关心线程模型的具体实现，即可对任务进行分发、广播等操作。
class IFoxThread {
public:typedef std::shared_ptr<IFoxThread> ptr;//被调度的任务typedef std::function<void()> callback;virtual ~IFoxThread() {};//将一个任务分发给某个线程(通常是随机或轮询方式选择)virtual bool dispatch(callback cb);//将任务分发到指定的ID的线程，支持线程定向调度virtual bool dispatch(uint32_t id, callback cb);//批量分发任务，通常用于高性能场景下一次性提交多个任务virtual bool batchDispatch(const std::vector<callback>& cbs);//广播任务，让所有线程都执行一次回调函数，如配置更新、通知等virtual void broadcast(callback cb);//启动线程或线程池virtual void start();//停止线程或线程池virtual void stop();//等待所有线程执行完成，用于程序退出前virtual void join();//输出线程状态到流virtual void dump(std::ostream& os);//返回线程池总共调度过的任务数量virtual uint64_t getTotal();
};//FoxThread 类是一个专门用于多线程任务调度的类，它通过事件驱动模型实现高效的任务处理。它通过 libevent 库的事件循环机制，
//能够在后台不断监听和处理任务。每个 FoxThread 实例都代表一个独立的工作线程，负责执行分发到该线程的任务。通过提供灵活的
//任务调度接口，FoxThread 允许外部线程将任务分发到当前线程，并确保任务的顺序执行和线程安全。
//实现上，FoxThread 利用了 UNIX socketpair 管道机制实现线程间的通信，通过 m_read 和 m_write 套接字监听任务的到来。
//当任务通过 dispatch() 或 batchDispatch() 等接口被派发时，FoxThread 会将这些任务加入到任务队列中，并通过 m_write 
//通知线程执行任务。在事件循环中，FoxThread 会持续等待来自 m_read 的可读事件，一旦有任务被写入，它就会触发回调函数并
//处理任务。通过这种方式，FoxThread 实现了任务的异步调度和高效处理，适用于高并发的多线程应用。
class FoxThread : public IFoxThread {
public:typedef std::shared_ptr<FoxThread> ptr;typedef IFoxThread::callback callback;//初始化回调函数的函数typedef std::function<void(FoxThread*)> init_cb;//name：该线程的名字//base：事件循环的上下文//事件循环：持续等待事件并处理事件的循环机制，本质是这样一种逻辑// while (running) {//     等待某个事件（网络、定时器、信号、通知等）；//     有事件发生就分发给对应的处理回调函数；// }// 在 libevent 中，这个循环是由 event_base_dispatch() 或 event_base_loop() 来运行的，背后由 event_base* 这个结构管理。FoxThread(const std::string& name, struct event_base* base = NULL);~FoxThread();//获取的当前线程关联的FoxThread实例static FoxThread* GetThis();static const std::string& GetFoxThreadName();//程序可能创建多个 FoxThread 实例，GetAllFoxThreadName() 允许开发者一次性获取所有线程的名称//存放threadId----threadNamestatic void GetAllFoxThreadName(std::map<uint64_t, std::string>& name);//将当前 FoxThread 实例绑定到线程局部存储中（例如 TLS）。void setThis();//解绑当前线程中的 FoxThread 实例。void unsetThis();bool dispatch(callback cb) override;bool dispatch(uint32_t id, callback cb) override;bool batchDispatch(const std::vector<callback>& cbs) override;void broadcast(callback cb) override;void start() override;void stop() override;void join() override;void dump(std::ostream& os) override;uint64_t getTotal() override;//检查线程是否已经启动bool isStart() const;//获取事件基础结构struct event_base* getBase();//获取当前线程的idstd::thread::id getId() const;//获取某个名字下绑定的私有数据void* getData(const std::string& name);template<typename T>T* getData(const std::string& name);//在线程中存储私有数据void setData(const std::string& name, void* v);//初始化回调void setInitCb(init_cb v);private://FoxThread的主线程循环函数，由std::thread启动时执行//thread_cb() 是 线程主函数，它负责：// 初始化 event_base（即事件循环上下文）；// 创建事件监听器（监听 m_read）；// 将 read_cb 注册为 m_read 的回调函数；// 启动事件循环（通过 event_base_dispatch()）；void thread_cb();//与事件系统绑定的回调函数，当m_read可读时(说明主线程发来了任务)触发//从任务队列中取出任务，并顺序执行//触发条件：//主线程通过 m_write 套接字写入任意内容时，会触发 m_read 的可读事件，从而执行 read_cb。static void read_cb(evutil_socket_t sock, short which, void* args);private://这两个成员是UNIX socketpair管道的两个端点，用于线程间通信evutil_socket_t m_read; //在FoxThread所在线程中监听此管道，一旦有数据读入，就唤醒event loop执行任务evutil_socket_t m_write;//在主线程或外部线程调用dispatch时使用，把任务写入通道//libevent的"事件循环"核心结构，代表一个事件多路复用的运行环境//负责// 管理所有注册的 event（比如监听 m_read）；// 负责阻塞等待 IO 事件并触发回调；// 类似于 epoll_wait() 背后的调度器。struct event_base* m_base;//代表绑定到m_read管道上的读事件// 会设置成“可读时回调 read_cb()”；// 一旦有任务被 dispatch() 派发到当前线程，就通过 m_write 写入一个字节唤醒这个事件。struct event* m_event;//该FoxThread实际运行的工作线程//当这个线程被启动时，它运行的主函数是thread_cb()std::thread* m_thread;//保护共享资源myproject::RWMutex m_mutex;//该任务队列来存放所有通过dispatch()等接口派发到该线程的回调函数std::list<callback> m_callbacks;//线程的名称std::string m_name;//初始化回调函数//在thread_cb启动前调用//通常用于给当前线程初始化上下文、设置线程局部变量等。init_cb m_initCb;//标记当前线程是否在处理任务的状态标志bool m_working;//标志线程是否已经启动bool m_start;//记录当前线程累计处理的任务数量，便于监控、统计用途。uint64_t m_total;//为当前FoxThread绑定一些自定义的数据对象std::map<std::string, void*> m_datas;
};//FoxThreadPool 类是一个多线程池的实现，旨在通过预先创建多个线程来高效地处理任务。它通过对多个 FoxThread 实例的管理，
// 使得任务能够被快速调度到空闲的线程中，从而提升了任务处理的效率。在任务调度时，FoxThreadPool 会根据当前的线程池状态，
// 决定是将任务随机分配到某个线程，还是指定某个线程来处理任务。同时，它还支持批量任务调度，能够一次性将多个任务分配给线程
// 池中的线程进行处理。
//在实现上，FoxThreadPool 维护了一个 FoxThread 对象的列表，并通过 m_freeFoxThreads 列表来管理空闲的线程。
// 任务被添加到一个等待队列中，当有线程空闲时，它会取出任务并执行。FoxThreadPool 还支持线程池的初始化回调，
// 使得每个线程在启动时执行特定的初始化操作。通过这些机制，FoxThreadPool 可以高效地复用线程、调度任务，
// 并在任务负载增加时动态处理任务分发。
class FoxThreadPool : public IFoxThread {
public:typedef std::shared_ptr<FoxThreadPool> ptr;typedef IFoxThread::callback callback; //size：线程池中要创建多少个线程//name：线程池的名字//advance：是否使用高级模式/调度增强模式FoxThreadPool(uint32_t size, const std::string& name = "", bool advance = false);~FoxThreadPool();//随机线程执行bool dispatch(callback cb) override;bool batchDispatch(const std::vector<callback>& cb) override;//指定线程执行bool dispatch(uint32_t id, callback cb) override;//从线程列表中随机获取一个线程FoxThread* getRandFoxThread();//指定线程启动时要执行的初始化动作void setInitCb(FoxThread::init_cb v) { m_initCb = v;}void start() override;void stop() override;void join() override;void dump(std::ostream& os) override;void broadcast(callback cb) override;uint64_t getTotal() override { return m_total;}private://回收线程void releaseFoxThread(FoxThread* t);//其作用通常是检查有没有空闲线程、是否该唤醒线程来处理等待队列中的任务。void check();// 包装任务回调，增加额外逻辑void wrapcb(std::shared_ptr<FoxThread>, callback cb);private://线程池中的线程数量，构造时指定uint32_t m_size;//当前分发用的线程索引，用于轮询分发任务uint32_t m_cur;//线程池的名字std::string m_name;bool m_advance;bool m_start;RWMutex m_mutex;//等待执行的任务列表，当前没有空闲线程时可临时存放任务std::list<callback> m_callbacks;std::vector<FoxThread*> m_threads;//当前空闲可调度的线程集合，用于高效复用线程std::list<FoxThread*> m_freeFoxThreads;//线程初始化时要执行的回调函数FoxThread::init_cb m_initCb;//已调度任务的总数uint64_t m_total;
};//FoxThreadManager 类是一个管理多个线程池和线程实例的中央管理器。它提供了对不同线程池的任务调度、状态管理和资源控制功能。
// 通过这个类，开发者可以方便地向指定线程池派发任务，不仅支持单个任务的分配，还支持批量任务和广播任务，极大地简化了多线程任
// 务管理的复杂度。此外，FoxThreadManager 还可以将线程池的状态信息打印出来，便于监控和调试系统的运行状态。它的核心作用是
// 将线程池和线程实例按照名字进行管理，并提供简单的接口进行操作。
//实现上，FoxThreadManager 使用一个 std::map 数据结构来维护线程池或线程对象与其对应的名字之间的映射关系。通过这种映射，
// 管理器可以快速定位到指定的线程池或线程实例，并对其进行任务调度、状态打印等操作。在初始化和启动时，管理器会启动所有注册
// 的线程池或线程对象，停止时则会释放资源。FoxThreadManager 作为一个全局单例，确保了全系统范围内的线程管理和任务调度的
// 统一性和高效性。
class FoxThreadManager {
public:typedef IFoxThread::callback callback;//向名为name的线程池中随机选择一个线程派发任务。void dispatch(const std::string& name, callback cb);//向名为name的线程池中的 指定线程 派发任务。void dispatch(const std::string& name, uint32_t id, callback cb);// 向名为name的线程池中 批量派发多个任务。void batchDispatch(const std::string& name, const std::vector<callback>& cbs);// 向指定名字的线程池中的 所有线程广播执行同一个任务回调。void broadcast(const std::string& name, callback cb);// 将所有线程池或线程对象的状态信息打印到输出流。void dumpFoxThreadStatus(std::ostream& os);// 初始化线程管理器void init();// 启动所有注册进来的线程池或线程对象。void start();//停止所有注册的线程池或线程对象，释放资源。void stop() ;// 获取指定名称的线程池/线程实例。IFoxThread::ptr get(const std::string& name);// 向管理器中注册一个线程池或线程对象。void add(const std::string& name, IFoxThread::ptr thr);private://维护一个名字到线程/线程池之间的映射std::map<std::string, IFoxThread::ptr> m_threads;
};typedef myproject::Singleton<FoxThreadManager> FoxThreadMgr;}#endif

threadPool.cpp

#include "fox_thread.h"
#include "myproject/config.h"
#include "myproject/log.h"
#include "myproject/util.h"
#include "myproject/macro.h"
#include "myproject/config.h"
#include <iomanip>namespace myproject {static myproject::Logger::ptr g_logger = myproject_LOG_NAME("system");static myproject::ConfigVar<std::map<std::string, std::map<std::string, std::string> > >::ptr g_thread_info_set= Config::Lookup("fox_thread", std::map<std::string, std::map<std::string, std::string> >(),"confg for thread");//全局静态变量，用于跨多个 FoxThread 实例共享数据
static RWMutex s_thread_mutex;
static std::map<uint64_t, std::string> s_thread_names;
//线程局部变量，每个线程都有自己独立一份s_thread指针
//我们可以通过FoxThread::GetThis()拿到当前线程的FoxThread*实例
thread_local FoxThread* s_thread = nullptr;void FoxThread::read_cb(evutil_socket_t sock, short which, void* args) {FoxThread* thread = static_cast<FoxThread*>(args);uint8_t cmd[4096];if (recv(sock, cmd, sizeof(cmd), 0) > 0) {std::list<callback> callbacks;RWMutex::WriteLock lock(thread->m_mutex);callbacks.swap(thread->m_callbacks);lock.unlock();thread->m_working = true;//auto：std::list<std::function<void()>>::iteratorfor (std::list<std::function<void()>>::iterator it = callbacks.begin(); it != callbacks.end(); ++it) {if (*it) {try {(*it)();}  catch (std::exception& ex) {myproject_LOG_ERROR(g_logger) << "exception:" << ex.what();} catch (const char* c) {myproject_LOG_ERROR(g_logger) << "exception:" << c;} catch (...) {myproject_LOG_ERROR(g_logger) << "uncatch exception";}} else {//如果回调为空，说明要停止线程//中断事件循环，让线程退出event_base_loopbreak(thread->m_base);thread->m_start = false;//清空线程局部存储thread->unsetThis();break;}}myproject::Atomic::addFetch(thread->m_total, callbacks.size());}
}FoxThread::FoxThread(const std::string& name, struct event_base* base):m_read(0),m_write(0),m_base(NULL),m_event(NULL),m_thread(NULL),m_name(name),m_working(false),m_start(false),m_total(0) {int fds[2];if(evutil_socketpair(AF_UNIX, SOCK_STREAM, 0, fds) == -1) {throw std::logic_error("thread init error");}//设置套接字为非阻塞// 通过设置 m_read 和 m_write 为非阻塞套接字，FoxThread 可以保持高效、// 响应迅速的事件循环。它不必在等待任务时被阻塞，而是可以继续执行其他任务// 或周期性检查是否有新的任务。这种设计使得 FoxThread 能够以事件驱动的方// 式处理大量任务，并确保不会因为 I/O 操作的延迟而影响到整体的线程性能。evutil_make_socket_nonblocking(fds[0]);evutil_make_socket_nonblocking(fds[1]);m_read = fds[0];m_write = fds[1];if (base) {//当传入外部event_base*，意味着当前线程通常是主线程，将直接使用FoxThread的事件循环//主线程负责初始化 FoxThreadManager 和线程池，启动和管理所有子线程的生命周期（创建、启动、停止、回收等）。//而start中的子线程用来通过 libevent 事件循环等待并处理分发过来的任务（callback），执行异步任务逻辑。m_base = base;//将当前FoxThread绑定到线程setThis();} else {m_base = event_base_new();}//创建事件对象：// 使用 event_new 函数为 m_read（接收端套接字）创建一个新的事件。// EV_READ 表示当套接字可读时触发该事件。// EV_PERSIST 表示事件持续存在，即在事件触发后仍然有效。// read_cb 是触发事件时调用的回调函数，this 作为参数传递给回调函数。m_event = event_new(m_base, m_read, EV_READ | EV_PERSIST, read_cb, this);//添加事件到事件循环event_add(m_event, NULL);
}void FoxThread::dump(std::ostream& os) {RWMutex::ReadLock lock(m_mutex);os << "[thread name=" << m_name<< " working=" << m_working<< " tasks=" << m_callbacks.size()<< " total=" << m_total<< "]" << std::endl;
}std::thread::id FoxThread::getId() const {if (m_thread) {return m_thread->get_id();}//返回一个默认构造的id对象return std::thread::id();
}void* FoxThread::getData(const std::string& name) {auto it = m_datas.find(name);return it == m_datas.end() ? nullptr : it->second;
}void  FoxThread::setData(const std::string& name, void* v) {//Mutex::WriteLock lock(m_mutex);m_datas[name] = v;
}FoxThread::~FoxThread() {if (m_read) {close(m_read);}if (m_write) {close(m_write);}stop();join();if (m_thread) {delete m_thread;}if (m_event) {event_free(m_event);}if (m_base) {event_base_free(m_base);}
}void FoxThread::start() {if (m_thread) {throw std::logic_error("FoxThread is running");}//start后会创建一个新的子线程//这个子线程的目的就是让它在后台一直处理异步事件//让这个线程进入一个“事件驱动循环”中，它阻塞的不是 I/O，//而是等待事件（如 socket 可读、超时、信号等）触发回调。m_thread = new std::thread(std::bind(&FoxThread::thread_cb, this));m_start = true;
}void FoxThread::thread_cb() {setThis();//为当前线程设置名称pthread_setname_np(pthread_self(), m_name.substr(0,15).c_str());if (m_initCb) {//如果设置了初始化回调m_initCb(this);m_initCb = nullptr;}//启动libevent的事件循环，监听和处理事件。该线程将进入阻塞状态，持续等待任//务事件的触发并进行处理，直到事件循环被外部中止（如停止线程时）。event_base_loop(m_base, 0);
}// 将一个回调任务添加到线程的任务队列中，并通过 socket 激活事件循环，使子线程去处理这个任务。
bool FoxThread::dispatch(callback cb) {RWMutex::WriteLock lock(m_mutex);m_callbacks.push_back(cb);lock.unlock();uint8_t cmd = 1;if (send(m_write, &cmd, sizeof(cmd), 0) <= 0) {return false;}return true;
}bool FoxThread::dispatch(uint32_t id, callback cb) {return dispatch(cb);
}bool FoxThread::batchDispatch(const std::vector<callback>& cbs) {RWMutex::WriteLock lock(m_mutex);for (auto i : cbs) {m_callbacks.push_back(i);}lock.unlock();uint8_t cmd = 1;if (send(m_write, &cmd, sizeof(cmd), 0) <= 0) {return false;}return true;
}void FoxThread::broadcast(callback cb) {dispatch(cb);
}void FoxThread::stop() {RWMutex::WriteLock lock(m_mutex);m_callbacks.push_back(nullptr);if (m_thread) {uint8_t cmd = 0;send(m_write, &cmd, sizeof(cmd), 0);}
}void FoxThread::join() {if (m_thread) {m_thread->join();//delete 会释放堆内存，但不会把指针本身清空，指针仍然保存着原先的地址。如果之后不小心访问这个指针//（例如再次使用 *m_thread），就会访问已被释放的内存，行为是未定义的（undefined behavior），非常危险。//设置为 NULL/nullptr 后，访问就变成空指针访问，可以被程序快速识别和拦截//（很多平台下会直接崩溃在访问空指针的位置），更容易调试。delete m_thread;m_thread = NULL;}
}FoxThread* FoxThread::GetThis() {return s_thread;
}void FoxThread::setThis() {m_name = m_name + "_" + std::to_string(myproject::GetThreadId());s_thread = this;RWMutex::WriteLock lock(s_thread_mutex);s_thread_names[myproject::GetThreadId()] = m_name;
}void FoxThread::unsetThis() {s_thread = nullptr;RWMutex::WriteLock lock(s_thread_mutex);s_thread_names.erase(myproject::GetThreadId());
}const std::string& FoxThread::GetFoxThreadName() {FoxThread* t = GetThis();if (t) {return t->m_name;}uint64_t tid = myproject::GetThreadId();do {RWMutex::ReadLock lock(s_thread_mutex);auto it = s_thread_names.find(tid);if(it != s_thread_names.end()) {return it->second;}} while(0);do {RWMutex::WriteLock lock(s_thread_mutex);s_thread_names[tid] = "UNNAME_" + std::to_string(tid);return s_thread_names[tid];} while (0);
}void FoxThread::GetAllFoxThreadName(std::map<uint64_t, std::string>& names) {RWMutex::WriteLock lock(s_thread_mutex);names.insert(s_thread_names.begin(), s_thread_names.end());
}FoxThreadPool::FoxThreadPool(uint32_t size, const std::string& name, bool advance):m_size(size),m_cur(0),m_name(name),m_advance(advance),m_start(false),m_total(0) {m_threads.resize(size);for (size_t i = 0; i < size; ++i) {FoxThread* t(new FoxThread(name + "_" + std::to_string(i)));m_threads[i] = t;}
}FoxThreadPool::~FoxThreadPool() {for(size_t i = 0; i < m_size; ++i) {delete m_threads[i];}
}void FoxThreadPool::start() {for(size_t i = 0; i < m_size; ++i) {m_threads[i]->setInitCb(m_initCb);m_threads[i]->start();m_freeFoxThreads.push_back(m_threads[i]);}if(m_initCb) {m_initCb = nullptr;}m_start = true;check();
}void FoxThreadPool::stop() {for(size_t i = 0; i < m_size; ++i) {m_threads[i]->stop();}m_start = false;
}void FoxThreadPool::join() {for(size_t i = 0; i < m_size; ++i) {m_threads[i]->join();}
}void FoxThreadPool::releaseFoxThread(FoxThread* t) {{RWMutex::WriteLock lock(m_mutex);m_freeFoxThreads.push_back(t);}check();
}bool FoxThreadPool::dispatch(callback cb) {{myproject::Atomic::addFetch(m_total, (uint64_t)1);RWMutex::WriteLock lock(m_mutex);if (!m_advance) {//如果不是高级调度模式，则直接将任务cb分发给某个线程，这是一种轮询调度策略，平均分到每个线程return m_threads[m_cur++ % m_size]->dispatch(cb);}//如果是高级调度模式，则先将任务放入等待队列中，等待空闲线程来拉取任务m_callbacks.push_back(cb);}check();return true;
}bool FoxThreadPool::batchDispatch(const std::vector<callback>& cbs) {myproject::Atomic::addFetch(m_total, cbs.size());RWMutex::WriteLock lock(m_mutex);if(!m_advance) {for(auto cb : cbs) {m_threads[m_cur++ % m_size]->dispatch(cb);}return true;}for(auto cb : cbs) {m_callbacks.push_back(cb);}lock.unlock();check();return true;
}void FoxThreadPool::wrapcb(std::shared_ptr<FoxThread> thr, callback cb) {cb();
}bool FoxThreadPool::dispatch(uint32_t id, callback cb) {myproject::Atomic::addFetch(m_total, (uint64_t)1);return m_threads[id % m_size]->dispatch(cb);
}FoxThread* FoxThreadPool::getRandFoxThread() {return m_threads[m_cur++ % m_size];
}void FoxThreadPool::broadcast(callback cb) {for(size_t i = 0; i < m_threads.size(); ++i) {m_threads[i]->dispatch(cb);}
}void FoxThreadPool::dump(std::ostream& os) {RWMutex::ReadLock lock(m_mutex);os << "[FoxThreadPool name = " << m_name << " thread_count = " << m_threads.size()<< " tasks = " << m_callbacks.size() << " total = " << m_total<< " advance = " << m_advance<< "]" << std::endl;for(size_t i = 0; i < m_threads.size(); ++i) {os << "    ";m_threads[i]->dump(os);}
}           // 检查是否有空闲线程和待处理的回调任务，
// 如果有，则将任务派发到线程中执行
void FoxThreadPool::check() {do {// 如果线程池尚未启动，直接退出if (!m_start) {break;}// 加写锁，确保对线程池内部数据结构的修改是线程安全的RWMutex::WriteLock lock(m_mutex);// 如果没有空闲线程或没有待处理的任务，直接退出if (m_freeFoxThreads.empty() || m_callbacks.empty()) {break;}// 从空闲线程列表中取出一个线程（注意用 shared_ptr 包装，带自定义 deleter）std::shared_ptr<FoxThread> thr(m_freeFoxThreads.front(), std::bind(&FoxThreadPool::releaseFoxThread, this, std::placeholders::_1));m_freeFoxThreads.pop_front();// 取出一个待执行的回调任务callback cb = m_callbacks.front();m_callbacks.pop_front();// 解锁，因为后续执行任务派发可能较慢，不希望锁持有太久lock.unlock();// 如果线程已经启动，则派发任务（wrapcb 用于包装用户回调）if (thr->isStart()) {thr->dispatch(std::bind(&FoxThreadPool::wrapcb, this, thr, cb));} else {// 如果线程尚未启动，则将任务重新放回任务队列RWMutex::WriteLock lock(m_mutex);m_callbacks.push_front(cb);}} while (true); // 保证整个过程在可处理条件下至少执行一次，否则立即退出
}IFoxThread::ptr FoxThreadManager::get(const std::string& name) {auto it = m_threads.find(name);return it == m_threads.end() ? nullptr : it->second;
}void FoxThreadManager::add(const std::string& name, IFoxThread::ptr thr) {m_threads[name] = thr;
}void FoxThreadManager::dispatch(const std::string& name, callback cb) {IFoxThread::ptr ti = get(name);myproject_ASSERT(ti);ti->dispatch(cb);
}void FoxThreadManager::dispatch(const std::string& name, uint32_t id, callback cb) {IFoxThread::ptr ti = get(name);myproject_ASSERT(ti);ti->dispatch(id, cb);
}void FoxThreadManager::batchDispatch(const std::string& name, const std::vector<callback>& cbs) {IFoxThread::ptr ti = get(name);myproject_ASSERT(ti);ti->batchDispatch(cbs);
}void FoxThreadManager::broadcast(const std::string& name, callback cb) {IFoxThread::ptr ti = get(name);myproject_ASSERT(ti);ti->broadcast(cb);
}void FoxThreadManager::dumpFoxThreadStatus(std::ostream& os) {os << "FoxThreadManager: " << std::endl;for(auto it = m_threads.begin();it != m_threads.end(); ++it) {it->second->dump(os);}os << "All FoxThreads:" << std::endl;std::map<uint64_t, std::string> names;FoxThread::GetAllFoxThreadName(names);for(auto it = names.begin();it != names.end(); ++it) {os << std::setw(30) << it->first<< ": " << it->second << std::endl;}
}void FoxThreadManager::init() {auto m = g_thread_info_set->getValue();for (auto i : m) {//当前线程池需要的线程数auto num = myproject::GetParamValue(i.second, "num", 0);//线程或线程池的名auto name = i.first;auto advance = myproject::GetParamValue(i.second, "advance", 0);if(num <= 0) {myproject_LOG_ERROR(g_logger) << "thread pool:" << name<< " num:" << num<< " advance:" << advance<< " invalid";continue;}if (num == 1) {m_threads[name] = FoxThread::ptr(new FoxThread(name));} else {m_threads[name] = FoxThreadPool::ptr(new FoxThreadPool(num, name, advance));}}
}void FoxThreadManager::start() {for(auto i : m_threads) {myproject_LOG_INFO(g_logger) << "thread: " << i.first << " start begin";i.second->start();myproject_LOG_INFO(g_logger) << "thread: " << i.first << " start end";}
}void FoxThreadManager::stop() {for(auto i : m_threads) {myproject_LOG_INFO(g_logger) << "thread: " << i.first << " stop begin";i.second->stop();myproject_LOG_INFO(g_logger) << "thread: " << i.first << " stop end";}for(auto i : m_threads) {myproject_LOG_INFO(g_logger) << "thread: " << i.first << " join begin";i.second->join();myproject_LOG_INFO(g_logger) << "thread: " << i.first << " join end";}
} }

FoxThread的实现原理

FoxThread 是基于 libevent 实现的事件驱动线程，核心通过 socketpair 建立主线程与子线程的通信管道，子线程运行事件循环等待 m_read 上的读事件。主线程调用 dispatch() 提交任务并写入 m_write 唤醒线程，触发 read_cb() 执行任务队列中的回调。任务通过 mutex 加锁保护，并支持中断事件循环退出线程。整体设计轻量、异步高效，适用于高并发任务调度场景。

             +-------------------------+|     FoxThread 构造     ||-------------------------|| 创建 socketpair         || 设置非阻塞              || 初始化 event_base       || 创建 read_cb 事件       |+-----------+-------------+|v+-------------------+| 调用 start()      | ← 外部线程调用（如主线程）+-------------------+|v+--------------------------+| 创建 std::thread 子线程 || 执行 thread_cb()        |+-----------+--------------+|v+-------------------------------------+| thread_cb():                        || 设置线程名 + 执行初始化回调（可选） || 启动 event_base_loop() 事件循环     |+----------------+--------------------+|| (阻塞等待事件)|+----------------v---------------------------+| 主线程调用 dispatch(cb) 向 m_callbacks 添加 || 并通过 m_write 写入 socket 激活事件         |+----------------+---------------------------+|v+-----------------------+| read_cb 被触发        | ← 事件触发点（套接字可读）|-----------------------|| 读取 socket           || 获取所有 callback     || 执行 callback()       || 统计执行次数          || 若 callback=null 停止 |+-----------+-----------+|v+--------------------------+| 线程退出准备              || event_base_loopbreak()   || unsetThis() 清理上下文   |+--------------------------+

在 FoxThread 的整个生命周期中，从构造到销毁，它主要围绕事件驱动和任务派发这两大核心进行设计。线程在构造时首先通过 socketpair 创建一对本地通信套接字 m_read 与 m_write，并设置为非阻塞模式。这对套接字的作用是：主线程向 m_write 写入数据，以唤醒子线程监听的 m_read，触发事件循环中注册的 read_cb 回调函数。

随后，FoxThread 通过 event_base_new() 创建一个事件基础对象，并将 read_cb 回调函数注册到 m_read 套接字的 EV_READ | EV_PERSIST 事件上。此时，如果尚未传入外部的 event_base，则线程将自行管理其事件循环。

调用 start() 方法后，会创建一个新的子线程，在线程内部调用 thread_cb() 函数。在该函数中，线程首先通过 setThis() 将自己注册为线程局部实例，并执行可选的初始化回调，然后正式进入 event_base_loop() 的事件循环状态，开始等待任务事件。

任务的提交通过 dispatch() 完成，主线程将任务（即 callback）加入线程的 m_callbacks 队列中，并通过向 m_write 写入一个字节数据，唤醒 m_read 的可读事件。子线程接收到唤醒信号后，执行 read_cb() 回调函数，从 m_callbacks 中取出所有任务，逐个执行，每个任务都用 try-catch 块进行异常保护，确保线程的健壮性。

如果某一次任务为 nullptr，表示线程需要退出，这时 read_cb() 会调用 event_base_loopbreak() 跳出事件循环，并调用 unsetThis() 清理线程本地注册信息。

最后，外部线程可通过 join() 等待子线程安全退出，并自动释放资源。

这一机制实现了一个轻量级、高并发、事件驱动的线程任务执行模型，线程通过 libevent 事件循环持续运行，任务派发通过异步 socket 通知，既避免了繁重的同步阻塞，又兼顾了任务隔离与线程可控性，非常适合网络服务器、定时任务调度器等异步高效场景。

FoxThreadPool 的实现原理

FoxThreadPool 是基于多个 FoxThread 线程的池，旨在高效地分配和管理线程任务。其核心设计是通过将任务分发到多个线程上并利用任务队列进行调度，从而提高任务的并发执行能力，适合高并发任务调度和分布式处理场景。

             +-------------------------+|     FoxThreadPool 构造 ||-------------------------|| 创建多个 FoxThread      || 存储线程池大小与状态     |+-----------+-------------+|v+-------------------+| 调用 start()      | ← 外部线程调用（如主线程）+-------------------+|v+------------------------+| 启动所有子线程         || 调用 start() 执行      |+------------------------+|v+----------------------------------------+| 调度任务：主线程调用 dispatch()      || 或 batchDispatch() 进行任务提交     || 将任务插入 m_callbacks 队列         |+-----------------+----------------------+|v+--------------------------+| 任务分发到空闲线程池中的线程 || (根据调度策略分发)          |+--------------------------+|v+-------------------------+| 子线程接收并执行回调函数 || 任务处理完毕后回调 wrapcb |+-------------------------+|v+----------------------------+| 任务执行完毕后，线程返回池 || 线程池进行检查与任务重分配 |+----------------------------+

FoxThreadPool 的设计使得多线程池能够高效地管理和调度任务，通过分发任务到空闲线程，并使用事件驱动模型，最大化线程资源的利用率。在高并发场景下，能够保证线程池中的每个线程都有任务可执行，同时提供了灵活的任务分发机制和线程池管理策略，适合需要大量并发任务调度的应用场景。

FoxThreadManager 的实现原理

FoxThreadManager 是一个线程管理器，负责管理和调度多个线程（或线程池）。它通过维护一个线程名到线程实例的映射，允许外部代码通过线程名对线程进行操作。它支持线程的添加、获取、任务调度、广播任务等功能，适用于需要动态管理多个线程或线程池的高并发应用。

             +-------------------------+|     FoxThreadManager 构造 ||-------------------------|| 创建线程池 m_threads    |+-----------+-------------+|v+-------------------+| 调用 init()       | ← 外部线程调用（初始化线程池配置）+-------------------+|v+-------------------------+| 初始化线程池            || 设置线程数、名称等      || 将线程添加到 m_threads |+-------------------------+|v+-----------------------+| 调用 start()          | ← 外部线程调用（启动所有线程）+-----------------------+|v+--------------------------+| 每个线程调用 start()     || 启动事件循环与任务执行   |+--------------------------+|v+------------------------------------------+| 调度任务：主线程调用 dispatch()        || 向特定线程添加任务并执行回调            |+----------------+-------------------------+|v+----------------------------+| 执行任务：线程执行回调函数 |+----------------------------+|v+---------------------------+| 任务执行完毕，线程返回池 |+---------------------------+|v+---------------------------------+| 调用 stop() 停止所有线程池    || 调用 join() 等待线程退出      |+---------------------------------+

FoxThreadManager 通过维护一个线程池的映射，提供了灵活的线程管理能力。它支持根据线程名称获取线程、向线程分发任务、批量分发任务、广播任务以及线程池状态的管理和输出。通过初始化、启动、停止、等待线程等操作，FoxThreadManager 提供了一个高效、易用的线程管理工具，非常适合多线程并发处理、任务调度和分配。