| 類別 | 職責 | 關鍵成員 |
|---|
| EventLoop | 事件循環、線程綁定、任務隊列 | Poller、TimerQueue、wakeupFd、pendingFunctors |
| Channel | 文件描述符事件封裝、回調綁定 | fd、events、revents、readCallback、writeCallback |
| Poller | I/O 多路復用抽象 (epoll/poll/kqueue) | pollFdList、ChannelMap、poll() |
EventLoop：每個線程一個事件循環
class EventLoop : noncopyable {
public:
using Functor = std::function;
EventLoop(); // 綁定當前線程 threadId_
~EventLoop();
void loop(); // 啟動事件循環
void quit(); // 退出循環
// 跨線程任務提交
void runInLoop(Functor cb); // 當前線程直接執行，其他線程進 queueInLoop
void queueInLoop(Functor cb); // 線程安全加入 pendingFunctors_
// 定時器
TimerId runAt(Timestamp, Functor);
TimerId runAfter(double delay, Functor);
TimerId runEvery(double interval, Functor);
// Channel 管理 (供 Channel 調用)
void updateChannel(Channel* channel);
void removeChannel(Channel* channel);
bool hasChannel(Channel* channel);
// 喚醒機制 (跨線程通知)
void wakeup(); // 向 eventfd 寫入 1
private:
void doPendingFunctors(); // 執行所有排隊任務
void handleRead(); // eventfd 讀回調，消費 wakeup 數據
const pid_t threadId_; // 綁定線程 ID
std::unique_ptr poller_;
std::unique_ptr timerQueue_;
int wakeupFd_; // eventfd
std::unique_ptr wakeupChannel_; // 監聽 wakeupFd 讀事件
ChannelList activeChannels_; // 就緒 Channel 列表
Channel* currentActiveChannel_; // 當前處理的 Channel
std::mutex mutex_;
std::vector pendingFunctors_; // 跨線程任務隊列
bool callingPendingFunctors_; // 防止重入
};
核心循環：loop()
void EventLoop::loop() {
assertInLoopThread();
looping_ = true;
while (!quit_) {
activeChannels_.clear();
// 1. I/O 多路復用等待事件
poller_->poll(kPollTimeMs, &activeChannels_);
// 2. 處理所有就緒 Channel
for (Channel* channel : activeChannels_) {
currentActiveChannel_ = channel;
channel->handleEvent(receiveTime); // 分發到各具體回調
}
currentActiveChannel_ = nullptr;
// 3. 執行跨線程提交的任務
doPendingFunctors();
}
looping_ = false;
}
跨線程任務提交：runInLoop / queueInLoop
void EventLoop::runInLoop(Functor cb) {
if (isInLoopThread()) {
cb(); // 同線程直接執行
} else {
queueInLoop(std::move(cb)); // 異線程入隊
}
}
void EventLoop::queueInLoop(Functor cb) {
{
std::lock_guard lock(mutex_);
pendingFunctors_.push_back(std::move(cb));
}
if (!isInLoopThread() || callingPendingFunctors_) {
wakeup(); // 喚醒 EventLoop 處理新任務
}
}
void EventLoop::wakeup() {
uint64_t one = 1;
ssize_t n = sockets::write(wakeupFd_, &one, sizeof(one));
// eventfd 語意：寫入非零值觸發可讀事件
}
Channel：文件描述符事件封裝
class Channel : noncopyable {
public:
using EventCallback = std::function;
using ReadEventCallback = std::function;
Channel(EventLoop* loop, int fd);
~Channel();
// 事件註冊
void setReadCallback(const ReadEventCallback& cb) { readCallback_ = cb; }
void setWriteCallback(const EventCallback& cb) { writeCallback_ = cb; }
void setCloseCallback(const EventCallback& cb) { closeCallback_ = cb; }
void setErrorCallback(const EventCallback& cb) { errorCallback_ = cb; }
// 向 Poller 註冊/修改/刪除事件
void enableReading() { events_ |= kReadEvent; update(); }
void disableReading() { events_ &= ~kReadEvent; update(); }
void enableWriting() { events_ |= kWriteEvent; update(); }
void disableWriting() { events_ &= ~kWriteEvent; update(); }
void disableAll() { events_ = kNoneEvent; update(); }
void remove() { disabled_ = true; remove(); }
int fd() const { return fd_; }
int events() const { return events_; }
void setRevents(int revt) { revents_ = revt; }
EventLoop* ownerLoop() const { return loop_; }
private:
void update() { loop_->updateChannel(this); }
void remove() { loop_->removeChannel(this); }
// 事件分發入口
void handleEvent(Timestamp receiveTime) {
eventHandling_ = true;
if ((revents_ & EPOLLHUP) && !(revents_ & EPOLLIN)) {
if (closeCallback_) closeCallback_();
}
if (revents_ & (EPOLLERR | EPOLLNVAL)) {
if (errorCallback_) errorCallback_();
}
if (revents_ & (EPOLLIN | EPOLLPRI | EPOLLRDHUP)) {
if (readCallback_) readCallback_(receiveTime);
}
if (revents_ & EPOLLOUT) {
if (writeCallback_) writeCallback_();
}
eventHandling_ = false;
}
EventLoop* loop_;
const int fd_;
int events_; // 關注的事件
int revents_; // 實際發生的事件
int index_; // Poller 內部索引
bool eventHandling_;
bool disabled_;
ReadEventCallback readCallback_;
EventCallback writeCallback_;
EventCallback closeCallback_;
EventCallback errorCallback_;
};
Poller：I/O 多路復用抽象層
class Poller : noncopyable {
public:
using ChannelList = std::vector;
Poller(EventLoop* loop);
virtual ~Poller();
// 核心介面
virtual Timestamp poll(int timeoutMs, ChannelList* activeChannels) = 0;
virtual void updateChannel(Channel* channel) = 0;
virtual void removeChannel(Channel* channel) = 0;
virtual bool hasChannel(Channel* channel) const = 0;
static Poller* newDefaultPoller(EventLoop* loop); // 工廠方法
protected:
using ChannelMap = std::unordered_map;
ChannelMap channels_; // fd -> Channel*
EventLoop* ownerLoop_;
};
// EPollPoller 實現 (Linux)
class EPollPoller : public Poller {
public:
EPollPoller(EventLoop* loop);
~EPollPoller();
Timestamp poll(int timeoutMs, ChannelList* activeChannels) override;
void updateChannel(Channel* channel) override;
void removeChannel(Channel* channel) override;
bool hasChannel(Channel* channel) const override;
private:
static const int kNew = -1;
static const int kAdded = 1;
static const int kDeleted = 2;
void fillActiveChannels(int numEvents, ChannelList* activeChannels);
void update(int operation, Channel* channel); // EPOLL_CTL_ADD/MOD/DEL
int epollfd_;
std::vector events_; // epoll_wait 輸出緩衝
};
epoll 核心操作
void EPollPoller::update(int operation, Channel* channel) {
epoll_event event;
event.events = channel->events();
event.data.ptr = channel;
int fd = channel->fd();
if (::epoll_ctl(epollfd_, operation, fd, &event) < 0) {
// 錯誤處理...
}
}
Timestamp EPollPoller::poll(int timeoutMs, ChannelList* activeChannels) {
int numEvents = ::epoll_wait(epollfd_, events_.data(), static_cast(events_.size()), timeoutMs);
Timestamp now(Timestamp::now());
if (numEvents > 0) {
fillActiveChannels(numEvents, activeChannels);
if (static_cast(numEvents) == events_.size()) {
events_.resize(events_.size() * 2); // 動態擴容
}
}
return now;
}
void EPollPoller::fillActiveChannels(int numEvents, ChannelList* activeChannels) {
for (int i = 0; i < numEvents; ++i) {
Channel* channel = static_cast(events_[i].data.ptr);
channel->setRevents(events_[i].events);
activeChannels->push_back(channel);
}
}
三大組件協作時序圖
註冊階段:
App → Channel::enableReading()
→ EventLoop::updateChannel()
→ Poller::updateChannel() → epoll_ctl(ADD)
事件循環:
EventLoop::loop()
→ Poller::poll() → epoll_wait()
→ fillActiveChannels() → activeChannels_
→ Channel::handleEvent() → readCallback_()
→ 用戶代碼 (如 Acceptor::handleRead)
跨線程喚醒:
OtherThread → EventLoop::queueInLoop(task)
→ pendingFunctors_.push_back()
→ EventLoop::wakeup() → write(eventfd)
→ EventLoop::loop() 下輪迴代
→ Poller::poll() 返回 (eventfd 可讀)
→ Channel::handleEvent() → handleRead()
→ read(eventfd) 消費數據
→ doPendingFunctors() → 執行 task
設計亮點總結
- 線程綁定斷言：`assertInLoopThread()` 到處佈防，編譯期/執行期雙重保障
- eventfd 跨線程通知：無鎖、高效、統一事件源 (同樣走 Poller)
- Channel 解耦：fd 與事件處理邏輯分離，Poller 不知業務細節
- 共享指標管理連線：`enable_shared_from_this` 保證回調中連線存活
- RAII 資源管理：Socket、Channel、Timer 自動清理
- 上一篇文章： muduo 網路庫源碼剖析 (二)：TcpServer、Acceptor 與 TcpConnection 核心類詳解
- 下一篇文章： 解析 malloc 記憶體對齊策略：為什麼 new 1 Byte 和 24 Byte 佔用相同記憶體？
Copyright ©
程式碼之家 All Rights Reserved
