企业官网建设流程全解析

主要涉及两个核心类：EventPoller和EventPollerPool。这两个类在实现高并发网络处理、I/O多路复用以及任务调度方面起着至关重要的作用。以下将详细解释这两个类的设计理念、功能以及内部机制。

1.EventPoller 类

目的与定位：

EventPoller是 ZLMediaKit 中用于管理和轮询I/O事件的核心组件。它负责监控文件描述符（如套接字）的各种事件（读、写、错误等），并在事件发生时触发相应的回调函数。此外，EventPoller还管理异步任务的调度与执行，支持延迟任务和定时任务，确保高效的事件处理和任务管理。

主要特性：

1. I/O 多路复用：

   • 在 Linux 系统上，EventPoller利用epoll实现高效的I/O多路复用，能够同时监控大量文件描述符的事件，大大提升了高并发场景下的性能。
   • 支持注册、修改和取消文件描述符上的读写、错误等事件，并通过回调机制处理这些事件。

2. 异步任务调度：

   • 提供异步任务队列，允许在事件循环中安全地添加和执行任务。
   • 支持任务的延迟执行，可以在指定的时间之后执行一次性任务或重复任务。

3. 内部线程管理：

   • EventPoller可以在独立的线程中运行事件循环，也可以在当前线程中阻塞运行，提供了灵活的线程管理方式。
   • 通过内部的管道机制，实现线程间的通信和任务调度，避免了复杂的锁机制，提高了效率。

4. 线程安全：

   • 使用互斥锁和信号量等同步机制，确保在多线程环境下对任务队列和事件注册的安全访问。
   • 支持跨线程调用，通过异步方式将任务安全地添加到任务队列中，由事件循环线程执行。

核心方法详解：

• addEvent(int fd, int event, PollEventCB cb)：

  • 功能：注册一个文件描述符 (fd)，监听指定的事件 (event)，并绑定相应的回调函数 (cb)。
  • 实现机制：
    • 如果当前在事件循环线程中，直接调用epoll_ctl将事件添加到epoll实例中。
    • 否则，通过异步任务的方式，将注册操作委托给事件循环线程执行，确保线程安全。

• delEvent(int fd, PollCompleteCB cb = nullptr)：

  • 功能：取消对指定文件描述符 (fd) 的事件监听，并可选地绑定一个完成回调 (cb)。
  • 实现机制：
    • 类似于addEvent，如果在事件循环线程中，直接调用epoll_ctl删除事件。
    • 否则，通过异步任务将删除操作委托给事件循环线程执行。

• modifyEvent(int fd, int event, PollCompleteCB cb = nullptr)：

  • 功能：修改已注册文件描述符 (fd) 的监听事件类型 (event)，并可选地绑定一个完成回调 (cb)。
  • 实现机制：
    • 在事件循环线程中，使用epoll_ctl修改事件类型。
    • 非事件循环线程时，通过异步任务方式委托执行。

• async(TaskIn task, bool may_sync = true)和async_first(TaskIn task, bool may_sync = true)：

  • 功能：将任务 (task) 添加到异步任务队列中执行。async_first将任务优先添加到队列前端，确保其优先执行。
  • 实现机制：
    • 如果调用线程是事件循环线程且may_sync参数为true，则任务直接同步执行。
    • 否则，将任务添加到任务队列，并通过内部管道写入一个字节，唤醒事件循环线程以处理新任务。

• doDelayTask(uint64_t delay_ms, std::function<uint64_t()> task)：

  • 功能：调度一个延迟任务，在指定的毫秒延迟 (delay_ms) 后执行。任务函数可以返回下一个延迟时间，实现重复执行。
  • 实现机制：
    • 计算任务的执行时间点 (now + delay_ms)，将任务插入到按时间排序的std::multimap中。
    • 在事件循环中定期检查并执行到期的任务，同时重新调度需要重复执行的任务。

• runLoop(bool blocked, bool ref_self)：

  • 功能：启动事件循环，可以选择在当前线程中阻塞运行或创建独立线程运行。同时，可选择是否将当前EventPoller实例关联到线程本地存储，以便通过getCurrentPoller获取。
  • 实现机制：
    • 如果blocked为false，则在独立线程中运行事件循环，并通过信号量同步线程启动。
    • 事件循环中使用epoll_wait监听事件，处理I/O事件和异步任务队列中的任务，管理延迟任务的执行。

内部机制深入解析：

1. 事件轮询与处理：

   • 使用epoll_wait阻塞等待事件的发生，超时时间根据最近的延迟任务决定。
   • 当事件发生时，遍历所有触发的事件，调用相应的回调函数。
   • 内部管道的读端也是被监听的文件描述符之一，用于接收来自其他线程的任务调度信号。通过写入管道，可以唤醒事件循环线程处理新任务。

2. 任务队列管理：

   • 任务队列由一个双向链表 (List<Task::Ptr> _list_task) 实现，受互斥锁 (std::mutex _mtx_task) 保护。
   • 任务通过async_l方法添加到队列中，并通过内部管道写入信号，唤醒事件循环线程及时处理新任务。
   • 在事件循环线程中，读取所有待执行的任务并逐一执行，确保任务的执行顺序和线程安全。

3. 延迟任务管理：

   • 延迟任务存储在一个多重映射 (std::multimap<uint64_t, DelayTask::Ptr> _delay_task_map) 中，按照任务的执行时间顺序排序。
   • -任务通过doDelayTask方法添加到map中。
   • 每次事件轮询时，计算下一个任务的延迟时间，设置epoll_wait的超时时间以最小化延迟。
   • 到期的任务被执行，并根据任务返回值决定是否重新调度（实现重复执行）。

4. 异步网络编程：

• 下面的代码实现了一个异步的Socket连接监听机制，利用EventPoller监控Socket的可写事件来判断连接是否成功

// 监听该socket是否可写，可写表明已经连接服务器成功intresult=strong_self->_poller->addEvent(sock,EventPoller::Event_Write|EventPoller::Event_Error,[weak_self,sock,con_cb](intevent){if(autostrong_self=weak_self.lock()){// socket可写事件，说明已经连接服务器成功strong_self->setSock(strong_self->makeSock(sock,SockNum::Sock_TCP));strong_self->onConnected(sock,con_cb);}else{CLOSE_SOCK(sock);}});

• socket建立连接后，onConnected利用attachEvent（），在Socket上注册事件监听，并处理不同类型的Socket事件。

boolSocket::attachEvent(intsock,SockNum::SockType type){weak_ptr<Socket>weak_self=shared_from_this();if(type==SockNum::Sock_TCP_Server){// tcp服务器autoresult=_poller->addEvent(sock,EventPoller::Event_Read|EventPoller::Event_Error,[weak_self,sock](intevent){// 对于服务器端的监听套接字（通常是通过listen函数创建的套接字），Event_Read事件表示有新的客户端连接请求到来if(autostrong_self=weak_self.lock()){strong_self->onAccept(sock,event);}});}// tcp客户端或udpautoread_buffer=_poller->getSharedBuffer();autoresult=_poller->addEvent(sock,EventPoller::Event_Read|EventPoller::Event_Error|EventPoller::Event_Write,[weak_self,sock,type,read_buffer](intevent){autostrong_self=weak_self.lock();if(event&EventPoller::Event_Read){strong_self->onRead(sock,type,read_buffer);}if(event&EventPoller::Event_Write){strong_self->onWriteAble(sock,type);}if(event&EventPoller::Event_Error){strong_self->emitErr(getSockErr(sock));}});}

6. 错误处理与恢复：

   • 在事件轮询过程中，如果内部管道发生错误（如被关闭），则重新打开管道并重新注册事件，确保事件循环的持续运行。
   • 捕捉并记录回调和任务执行过程中抛出的异常，防止单个任务的异常导致整个事件循环崩溃。

7. 资源管理：

   • 通过智能指针 (std::shared_ptr和std::weak_ptr) 管理资源的生命周期，确保事件回调和任务执行结束后资源被正确释放。
   • 在析构函数中，调用shutdown方法确保事件循环线程安全退出，并关闭epoll文件描述符和内部管道。

设计模式与原则：

• 单例模式：

  • 通过EventPoller::Instance()方法获取EventPollerPool中的第一个EventPoller实例，确保全局只有一个默认的轮询器实例。

• 观察者模式：

  • 使用回调函数机制监听和响应I/O事件，实现事件驱动的编程模型。

• RAII（资源获取即初始化）：

  • 在构造函数和析构函数中管理资源的分配与释放，确保资源的正确生命周期管理，避免内存泄漏和资源竞争。

• 线程安全：

  • 通过互斥锁和信号量等同步机制，确保在多线程环境下对共享资源（如任务队列和事件注册）的安全访问。

2.EventPollerPool 类

目的与定位：

EventPollerPool负责管理一组EventPoller实例，构建一个高效的轮询器池，实现负载均衡和高并发处理能力。通过将工作负载分配到多个EventPoller，EventPollerPool能够充分利用多核处理器资源，提升系统的整体并发处理能力和性能。

主要特性：

1. 轮询器池化管理：

   • 维护多个EventPoller实例，根据配置的池大小初始化相应数量的轮询器，通常与系统的CPU核心数匹配，以优化并行处理能力。

2. 负载均衡与调度策略：

   • 提供机制根据轮询器的负载情况选择最合适的EventPoller，确保工作负载均衡分配，避免单个轮询器过载。

3. 线程亲和性优化（CPU Affinity）：

   • 通过enableCpuAffinity方法支持将不同的EventPoller绑定到特定的CPU核心，提升缓存命中率，减少上下文切换开销，提高性能。

4. 灵活的轮询器获取：

   • 提供getPoller方法，支持优先返回当前线程关联的EventPoller，以提升线程亲和性和减少跨线程访问的开销。
   • 如果需要均衡负载，可以忽略当前线程的轮询器，选择负载较轻的轮询器来处理任务。

5. 单例模式：

   • 通过EventPollerPool::Instance()方法获取全局唯一的EventPollerPool实例，确保整个应用程序中只有一个轮询器池。

核心方法详解：

• getPoller(bool prefer_current_thread = true)：

  • 功能：获取一个EventPoller实例来处理任务。参数prefer_current_thread决定是否优先返回当前线程关联的EventPoller。
  • 实现机制：
    • 如果prefer_current_thread为true，且当前线程已关联一个EventPoller，则返回该实例。
    • 否则，基于轮询器的负载情况，选择一个负载较轻的EventPoller实例返回。

• getFirstPoller()：

  • 功能：返回轮询器池中的第一个EventPoller实例，主要用于兼容旧代码或需要特定操作的场景。

• setPoolSize(size_t size = 0)：

  • 功能：设置轮询器池的大小（即EventPoller实例的数量）。在EventPollerPool单例实例创建之前调用有效。
  • 实现机制：
    • 如果size为0，则默认为系统的硬件并发数（CPU核心数）。
    • 根据设置的大小创建相应数量的EventPoller实例，并启动各自的事件循环。

• enableCpuAffinity(bool enable)：

  • 功能：设置是否启用EventPoller线程的CPU亲和性。开启后，每个EventPoller线程将绑定到特定的CPU核心，以优化性能。
  • 实现机制：
    • 在创建EventPoller实例时，根据设置决定是否调用系统API（如pthread_setaffinity_np）将线程绑定到指定CPU。

• preferCurrentThread(bool flag = true)：

  • 功能：配置getPoller方法是否优先返回当前线程关联的EventPoller。
  • 实现机制：
    • 设置内部标志_prefer_current_thread，影响getPoller的行为。

内部机制深入解析：

1. 轮询器实例管理：

   • EventPollerPool通过继承自TaskExecutorGetterImp（基类）来管理多个EventPoller实例。
   • 在构造函数中，根据配置的池大小创建EventPoller实例，并启动它们的事件循环。
   • 确保所有EventPoller实例正确初始化，并在应用程序退出时安全关闭。

2. 负载监测与调度：

   • 维护每个EventPoller的负载信息，通过接口（如getExecutorLoad）实时获取每个轮询器的当前负载情况。
   • 在调用getPoller时，根据负载信息选择最适合的EventPoller，确保任务均衡分配，避免热点轮询器成为性能瓶颈。

3. 线程亲和性与CPU优化：

   • 启用CPU亲和性后，每个EventPoller线程被绑定到特定的CPU核心，利用CPU缓存，提高执行效率。
   • 通过减少跨核缓存失效和上下文切换，显著提升高并发场景下的性能。

4. 事件通知与回调机制：

   • 通过信号中心 (NoticeCenter)，在轮询器池初始化完成后，发送启动完成的事件通知，允许其他组件进行相应的初始化或操作。

5. 资源管理与清理：

   • 在EventPollerPool的析构函数中，确保所有EventPoller实例的事件循环线程被安全关闭，并释放相关资源。

设计模式与原则：

• 单例模式：

  • EventPollerPool实现为单例模式，通过Instance方法提供全局唯一的访问点，避免多个池实例导致资源竞争和管理复杂性。

• 工厂模式：

  • 通过内部方法（如addPoller）动态创建和管理EventPoller实例，隐藏具体实现细节，对外提供统一的接口。

• 负载均衡策略：

  • 基于实时负载信息动态选择EventPoller，提升系统并发处理能力和响应速度。

• 高内聚低耦合：

  • EventPollerPool专注于管理EventPoller实例与任务分配，封装内部实现，降低与其他组件的耦合度，提高模块化和可维护性。

3.EventPoller 与 EventPollerPool 的协同工作机制

在 ZLMediaKit 中，EventPoller和EventPollerPool共同构建了一个高效、可扩展的事件处理和任务调度框架。其协同工作机制如下：

1. 初始化：
   • 在应用程序启动时，EventPollerPool根据配置（如池大小、CPU亲和性）初始化多个EventPoller实例。
   • 每个EventPoller运行在独立的线程中，负责监控和处理其注册的I/O事件，以及管理其所属的任务队列。

size_tTaskExecutorGetterImp::addPoller(conststring&name,size_t size,intpriority,boolregister_thread,boolenable_cpu_affinity){autocpus=thread::hardware_concurrency();size=size>0?size:cpus;for(size_t i=0;i<size;++i){autofull_name=name+" "+to_string(i);autocpu_index=i%cpus;EventPoller::Ptrpoller(newEventPoller(full_name));// 创建线程来运行事件循环poller->runLoop(false,register_thread);poller->async([cpu_index,full_name,priority,enable_cpu_affinity](){// 设置线程优先级ThreadPool::setPriority((ThreadPool::Priority)priority);// 设置线程名setThreadName(full_name.data());// 设置cpu亲和性if(enable_cpu_affinity){setThreadAffinity(cpu_index);}});_threads.emplace_back(std::move(poller));}returnsize;}

2. 任务分配与执行：

   • 当需要处理新的I/O事件或执行异步任务时，应用程序通过EventPollerPool::getPoller获取一个EventPoller实例。
   • 根据负载均衡策略，EventPollerPool选择一个负载较轻的EventPoller，将任务分配给它。
   • 选中的EventPoller执行任务，包括立即执行和延迟执行，确保任务的高效处理。

3. 事件处理：

   • 每个EventPoller持续监听其注册的文件描述符的I/O事件，如读写事件。
   • 当事件发生时，epoll捕捉到事件，并在事件循环中触发相应的回调函数，处理具体的I/O操作（如读取数据、写入数据等）。

4. 延迟任务与定时器：

   • EventPoller内部管理延迟任务，通过时间排序的std::multimap进行调度。
   • 定期检查当前时间与任务执行时间的差异，执行到期的任务，并根据任务的返回值决定是否重新调度，实现重复执行。

5. 线程安全与同步机制：

   • 通过互斥锁和信号量等同步机制，确保在多线程环境下对共享资源（如任务队列和事件注册）的安全访问。
   • 内部管道机制用于线程间通信，确保任务的及时调度和执行，避免复杂的锁机制导致的性能瓶颈。

6. 资源管理与异常处理：

   • 通过智能指针和RAII原则，自动管理资源的生命周期，确保资源的正确释放和防止内存泄漏。
   • 捕捉和记录回调和任务执行过程中的异常，防止单个任务的异常导致整个事件循环崩溃，提高系统的稳定性和可靠性。

4.设计理念与高并发实现原理

高效的I/O事件管理：

• 通过使用epoll实现高效的I/O多路复用，EventPoller能够同时监控大量文件描述符的事件，适用于高并发网络应用，如Web服务器、媒体服务器等。
• epoll的边缘触发和水平触发机制结合使用，优化事件通知和处理的效率，减少不必要的系统调用开销。

任务调度与并行处理：

• 异步任务队列和延迟任务管理使得EventPoller能够在事件循环中处理多种类型的任务，实现任务的高效调度和执行。
• 通过EventPollerPool的池化管理，将任务分配到多个EventPoller实例，充分利用多核处理器，提高系统的并行处理能力，降低单个线程的负载。

线程管理与CPU亲和性：

• 每个EventPoller运行在独立的线程中，避免了单个线程成为性能瓶颈。
• 可选的CPU亲和性设置将轮询器线程绑定到特定CPU核心，优化缓存利用率，减少跨核访问的开销，提升高并发场景下的性能。

同步与线程安全：

• 通过使用互斥锁和内部管道机制，确保在多线程环境下的任务调度和事件处理的线程安全性。
• 避免了复杂的锁竞争和潜在的死锁问题，通过异步任务调度和事件驱动模型，提高了系统的响应速度和稳定性。

资源管理与异常处理：

• 利用智能指针和RAII原则，自动管理资源的分配与释放，防止内存泄漏和资源泄露。
• 捕捉回调和任务执行过程中抛出的异常，确保事件循环的持续运行，提高系统的鲁棒性。

5.总结

EventPoller和EventPollerPool是 ZLMediaKit 实现高并发网络处理的核心组件。通过结合高效的I/O多路复用机制（如epoll）、灵活的任务调度与管理、线程亲和性优化以及严格的线程安全机制，这两个类共同构建了一个高性能、可扩展且稳定的事件处理与任务调度框架。

这种设计不仅能够应对大量并发连接和高频率的I/O事件，还能在复杂的任务调度和执行场景中保持高效和稳定，为开发高性能网络应用（如实时媒体服务器、Web服务器等）提供了坚实的基础。

理解和掌握EventPoller与EventPollerPool的内部机制和设计理念，对于开发者来说，不仅能够帮助他们更好地使用 ZLMediaKit，还能为开发定制化的高并发网络应用提供宝贵的参考和借鉴。