muduo库剖析(2)

muduo库剖析(2)

项目地址：https://github.com/huerni/cgmuduo.git

muduo是一个基于非阻塞IO和事件驱动的C++网络库，采用Reactor模式和one loop per thread + thread pool线程模型。

类功能介绍

Channel.*，Poller.*，EventLoop.* 三类为一体，在底层负责事件循环。EventLoop包含Channel，Poller，EventLoop负责轮询访问Poller，得到激活Channel列表，使Channel自己根据自身情况调用相应回调。

Thread.*，EventLoopThread.*，EventLoopThreadPool.*将线程与EventLoop绑定在一起，实现one loop per thread + thread pool线程模型。

TcpServer.*，Acceptor.*，TcpConnection.*为上层Reactor模式的实现。TcpServer面向用户，可由用户自定义回调方法以及启动服务器。Acceptor也看做分发器，当有新连接到来时，用来将连接转发给子线程处理。TcpConnection在子线程处理被分来的连接的具体事件。

执行流程

1. 用户创建TcpServer对象调用start()开启服务器，启动线程池，启动一个baseLoop(主线程)，由runInLoop()处理Acceptor(Reactor)的listen()方法，进行监听操作；
2. 一旦有客户端请求连接，Accetor调用TcpServer设置的newConnectionCallback_，将请求分发给subLoop(子线程)，即从线程池中取出，然后将连接丢给子线程，由TcpConnection处理，此后该连接的所有操作都由该子线程独自处理，与其他线程无关。
3. TcpConnection将读，写，关闭，错误操作与回调填充至绑定的Channel，然后调用runInLoop()和queueInLoop()由底层事件循环处理。
4. 底层事件循环中，主线程称为baseLoop，只执行监听操作与用户回调。而子线程具体执行操作由Channel具体返回的事件所执行。由图所示，蓝色部分由三类组成，EventLoop，Channel和Poller。每个EventLoop分别与各自的线程绑定，同时将Poller和Channel连接起来，调用loop()方法将Channel的结果送入Poller执行，将Poller返回的结果各自送入Channel执行。

Socket类

Socket类是对socket套接字的一层封装。

class Socket : noncopyable{
public:
    explicit Socket(int sockfd);
    ~Socket();

    int fd() const { return sockfd_; }
    void bindAddress(const InetAddress &localaddr); // 绑定地址
    void listen();  // 监听函数
    int accept(InetAddress *peeraddr); // 接收客户端
    void shutdownWrite();  // 关闭写端
	
    // 设置套接字属性
    void setTcpNoDelay(bool on);  // TCP_NODELAY  控制是否开启Nagle算法
    void setReuseAddr(bool on);  // SO_REUSEADDR  地址是否可重用
    void setReusePort(bool on);  // SO_REUSEPORT  允许将多个socket绑定到相同的地址和端口
    void setKeepAlive(bool on);  // SO_KEEPALIVE  用于开启或者关闭保活探测
private:
    const int sockfd_; // 套接字fd
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

Buffer类

Buffer类为自定义缓冲区类，作为读写缓冲区。

buffer分三个部分，预留空间，可读区，可写区。

使用vector数据结构作为缓冲区，因为vector长度可变，且随机下标复杂度低，比较灵活。使用两个int索引来维护读位置和写位置。

为什么使用int索引而不是迭代器或指针来维护？

一旦vector扩容，迭代器会失效。

为了解决粘包问题，添加了8字节的预留空间。当收到网络信息时，可以计算长度并写入到预留空间中。

class Buffer{
public:
    static const size_t kCheapPrepend = 8;  // 预留空间大小
    static const size_t kInitalSize = 1024;  // buffer初始大小
    
    explicit Buffer(size_t initialSize = kInitalSize) : buffer_(kCheapPrepend+initialSize)
            , readerIndex_(kCheapPrepend)
            , writerIndex_(kCheapPrepend) 
    {}

    // 可读数据大小，即缓存区内已存数据大小
    size_t readableBytes() const;
    // 可写数据大小，即缓冲区剩余空间
    size_t writableBytes() const;
    // 可读位置之前的大小(包含预留空间+已读空间)
    size_t prependableBytes() const;
    // 可读数据起始地址
    const char* peek() const;
    // 读取后调整读写指针
    void retrieve(size_t len);
    // 将读写指针归为初始
    void retrieveAll();
    // 将缓存区内所有数据转为string
    std::string retrieveAllAsString();
    // 将指定长度可读数据转为string
    std::string retrieveAsString(size_t len);
    // 确保是否有足够空间可写，如果没有，扩充空间
    void ensureWriteableBytes(size_t len);
    // 将len长度data写入缓存区
    void append(const char* data, size_t len);
    // 可写数据起始地址
    char* beginWrite();
    const char* beginWrite() const;
    // 从fd上读数据
    ssize_t readFd(int fd, int* saveErrno);
    // 写入数据到fd中
    ssize_t writeFd(int fd, int* saveErrno);

private:
    char* begin(); // buffer起始地址
    const char* begin() const;
    // 扩充空间
    void makeSpace(size_t len);
    
    std::vector buffer_;  // 使用vector作为缓冲区
    size_t readerIndex_;  // 读位置
    size_t writerIndex_; // 写位置
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

从fd中读数据时，需要判断两种情况：

1. 缓冲区可写空间足够，所以可以将数据全部读入缓冲区
2. 可写空间不足，巧妙的办法是申请一段栈内缓冲区放入额外的数据，这样只需要调用一次readv函数就能得到全部数据，减少了IO开销。然后将数据从栈内缓冲区转移到buffer中。

ssize_t Buffer::readFd(int fd, int* saveErrno) {
    char extraBuffer[65536] = {0};
    struct iovec vec[2];

    const size_t writable = writableBytes();
    vec[0].iov_base = begin() + writerIndex_;
    vec[0].iov_len = writable;

    vec[1].iov_base = extraBuffer;
    vec[1].iov_len = sizeof extraBuffer;

    const int iovcnt = (writable < sizeof extraBuffer) ? 2 : 1;
    const ssize_t n = ::readv(fd, vec, iovcnt);
    if(n < 0) {
        *saveErrno = errno;
    }
    else if(n <= writable) {
        writerIndex_ += n;
    }
    else {
        writerIndex_ = buffer_.size();
        // 此时会扩充空间
        append(extraBuffer, n - writable);
    }

    return n;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

写入数据到fd中，只需要将buffer中所有可读数据发送出去即可

ssize_t Buffer::writeFd(int fd, int* saveErrno) {

    ssize_t n = ::write(fd, peek(), readableBytes());
    if(n < 0) {
        saveErrno = &errno;
    }

    return n;
}
1
2
3
4
5
6
7
8
9

当调用append空间不足时，会调用makeSpace扩充空间。

void makeSpace(size_t len) {
    // 当可写空间+已读空间不足时，扩充vector长度
    // (因为可读位置每次读取是向后走的，这样前面会有一段不可用但空白区域，类似线性队列)
    if(writableBytes() + prependableBytes() < len + kCheapPrepend) {
        // 可写大小只扩充到len
        buffer_.resize(writerIndex_ + len);
    }
    // 否则不需要扩充，只要将可读数据向前移动即可。
    else {
        size_t readable = readableBytes();
        std::copy(begin() + readerIndex_, begin() + writerIndex_, begin() + kCheapPrepend);
        readerIndex_ = kCheapPrepend;
        writerIndex_ = readerIndex_ + readable;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Channel类

Channel封装了socket对应的event，并设置了一系列修改event的回调函数。

muduo中，Channel只设置两种事件，即读事件和写事件，包含三种状态，分为是否可读，是否可写，是否关注事件。

epoll有自己关注的一系列事件，根据Channel所对应的event进行操作或对Channel对应的event进行修改。所以，为了方便poller对Channel进行处理，Channel必须包含更新自身状态，返回自身状态，根据自身状态处理对应的回调一系列方法。

class Channel : noncopyable {
public:
    // 回调函数
    using EventCallback = std::function;
    using ReadEventCallback = std::function;

    Channel(EventLoop* loop, int fd);
    ~Channel();
    
    // 处理事件的接口
    void HandleEvent(Timestamp receiveTime);

    // 设置回调函数，有读，写，关闭，错误事件
    void setReadCallback(ReadEventCallback cb);
    void setWriteCallback(EventCallback cb);
    void setCloseCallback(EventCallback cb);
    void setErrorCallback(EventCallback cb);

    // 绑定指针，将弱指针转为共享指针
    // 防止channel被手动关闭时，channel还在执行回调函数
    void tie(const std::shared_ptr& );
	
    // 返回socket对应的event状态
    int fd() const { return fd_; }
    int events() const { return events_; }
    // 更新event
    void set_revents(int revt) { revents_ = revt; }
    

    // 更新fd感兴趣的事件events
    void enableReading();
    void disableReading();
    void enableWriting();
    void disableWriting();
    void disableAll();
	
    // 判断socket对应event种类
    bool isNoneEvent() const;
    bool isWriteEvent() const;
    bool inReadEvent() const;

    int index() { return index_; }
    void set_index(int idx) { index_ = idx; }
    
    // 返回所属于的loop
    EventLoop* ownerLoop() { return loop_; }
    // 从loop中删除event
    void remove();

private:
    // 通知所属的eventloop更新events
    void update();
    // 根据事件event类型进行相应的处理
    void handleEventWithGuard(Timestamp receiveTime);

    // 三种事件
    static const int KNoneEvent;
    static const int KReadEvent;
    static const int KWriteEvent;
	
    // muduo采用one loop per thread策略，所以一个Channel自始至终只属于一个eventLoop类，且只属于一个IO线程。
    EventLoop *loop_; // 事件循环
    
    const int fd_; // 绑定的fd，poller监听的对象
    int events_;  // 注册fd感兴趣的事件
    int revents_;  // poller返回的具体发生的事件
    int index_;

    std::weak_ptr tie_;
    bool tied_;

    // 四种回调函数
    // channel通道里面能够获知fd最终发生的具体的事件events，所以它负责调用具体的回调函数
    ReadEventCallback readCallback_;
    EventCallback writeCallback_;
    EventCallback closeCallback_;
    EventCallback errorCallback_;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

当poller返回激活的Channel时，该Channel必须根据自身的状态进行对应的用户回调。

void Channel::handleEventWithGuard(Timestamp receiveTime) {
    // 对方已关闭，调用关闭回调
    if((revents_ & EPOLLHUP) && !(revents_ & EPOLLIN)) {
        if(closeCallback_) {
            closeCallback_();
        }
    }
	// 发生异常，调用异常回调
    if(revents_ & EPOLLERR) {
        if(errorCallback_) {
            errorCallback_();
        }
    }
	// 读事件，调用读取回调
    if(revents_ & (EPOLLIN | EPOLLPRI)) {
        if(readCallback_) {
            readCallback_(receiveTime);
        }
    }
	// 写事件，调用写回调
    if(revents_ & EPOLLOUT) {
        if(writeCallback_) {
            writeCallback_();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

Poller类

Poller是一个虚函数，包含了IO复用方法的一些基本接口，可由IO复用方法进行继承扩展。

各种IO复用方法可查看：cg网络编程记录-IO复用

class Poller : noncopyable {
public:
    using ChannelList = std::vector;

    Poller(EventLoop* loop);
    virtual ~Poller() = default;

    // poller统一接口，包括操作，更新，删除，由继承实现
    virtual Timestamp poll (int timeoutMs, ChannelList *activeChannels) = 0;
    virtual void updateChannel(Channel* channel) = 0;
    virtual void removeChannel(Channel* channel) = 0;

    // 判断通道是否在该IO复用中
    bool hasChannel(Channel* channel) const;

    // 获取loop默认IO复用具体实现
    static Poller* newDefaultPoller(EventLoop* loop);

protected:
    // fd => channle实例
    using ChannelMap = std::unordered_map;
    // poller监视的所有channel
    ChannelMap channels_;
private:
    EventLoop* ownerloop_; // 所属的事件训练loop
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

Epoller类

继承自Poller类，采用epoll方式对socket进行监听。

epoll方式中，每个epoll都对应一个epollfd。首先要调用epoll_create创建epoll，获取epollfd，后续通过epollfd对该epoll进行操作。然后可以通过cpoll_ctl对指定的socket进行操作，如add/mod/del (muduo中由Epoller类对Channel类进行操作)，添加需要监听的Channel后，调用epoll_wait开始监听，返回被激活的Channel，然后处理事件。

class EpollPoller : public Poller {
public:
    EpollPoller(EventLoop* loop);
    ~EpollPoller() override;

    // 重写基类接口
    Timestamp poll (int timeoutMs, ChannelList *activeChannels) override;
    void updateChannel(Channel* channel) override;
    void removeChannel(Channel* channel) override;

private:
    static const int kInitEventListSize = 16;

    // 填充活跃通道
    void fillActiveChannels(int numEvents, ChannelList* activeChannels) const ;
    // 指定更新通道
    void update(int operation, Channel *channel);

    using EventList = std::vector;
	
    int epollfd_;
    EventList events_;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

epoll_create在构造函数中调用

EpollPoller::EpollPoller(EventLoop* loop) 
    : Poller(loop)
    , epollfd_(::epoll_create1(EPOLL_CLOEXEC))
    , events_(kInitEventListSize) { 
        if(epollfd_ < 0) {
            LOG_FATAL("epoll_create error:%d \n", errno);
        }
}
1
2
3
4
5
6
7
8

epoll_wait封转在update方法中。

// 指定操作更新Channel epoll_ctl具体操作  add/mod/del
void EpollPoller::update(int operation, Channel *channel) {
    int fd = channel->fd();
    epoll_event event; // epoll常规操作，将所需填入event结构中。
    bzero(&event, sizeof event);
    event.events = channel->events(); // channel感兴趣的事件
    event.data.ptr = (void*)channel;
    event.data.fd = fd;

    if(::epoll_ctl(epollfd_, operation, fd, &event) < 0) {
        if(operation == EPOLL_CTL_DEL) {
            LOG_ERROR("epoll_ctl del error: %d \n", errno);
        }
        else {
            LOG_FATAL("epoll_ctl add/mod error: %d \n", errno);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

Epoller类保存所有监听的Channel类，所以在更新Channel时，需要一些额外的操作。

void EpollPoller::updateChannel(Channel* channel) {
    // 获取Channel在poller的状态  未监听/已监听/已删除
    const int index = channel->index();
    // 如果poller面对一个新的Channel或者已经删除的Channel
    if(index == KNew || index == KDeleted) {
        if(index == KNew) {
            int fd = channel->fd();
            channels_[fd] = channel;
        }

        channel->set_index(KAdded);
        update(EPOLL_CTL_ADD, channel);
    }
    // Channel在poller的监听列表中
    else {
        int fd = channel->fd();
        // 如果Channel已经对任何事件不感兴趣，则把它剔除
        if(channel->isNoneEvent()) {
            update(EPOLL_CTL_DEL, channel);
            channel->set_index(KDeleted);
        }
        // 否则，直接更新它的状态
        else {
            update(EPOLL_CTL_MOD, channel);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

Epoller类还应该具备手动剔除Channel功能。

void EpollPoller::removeChannel(Channel* channel) {
    int fd = channel->fd();
    int index = channel->index();
    
    if(index == KAdded) {
        update(EPOLL_CTL_DEL, channel);
    }
    channels_.erase(fd);
    channel->set_index(KNew);
}
1
2
3
4
5
6
7
8
9
10

设置好以后，Epoll类调用poll开始监听，即封装epoll_wati

Timestamp EpollPoller::poll (int timeoutMs, ChannelList *activeChannels) {
    
    // epoll_wait返回被激活的socket
    int numEvents = ::epoll_wait(epollfd_, &*events_.begin(), static_cast(events_.size()), timeoutMs);
    int saveErrno = errno;
    Timestamp now(Timestamp::now());
	
    // 返回的新的事件填充至对应的Channel
    if(numEvents > 0) {
        LOG_INFO("%d events happened \n", numEvents);
        fillActiveChannels(numEvents, activeChannels);
        if(numEvents == events_.size()) {
            events_.resize(numEvents * 2);
        }
    }
    else if(numEvents == 0) {
        LOG_INFO("%s timeout! \n", __FUNCTION__);
    }
    else {
        if(saveErrno != EINTR) {
            errno = saveErrno;
            LOG_ERROR("EpollPoller::poll() err!");
        }
    }

    return now;
}

void EpollPoller::fillActiveChannels(int numEvents, ChannelList* activeChannels) const {
    // 修改被激活的Channel的状态
    for(int i = 0; i(events_[i].data.ptr);
        LOG_INFO("channel: %d", channel->fd());
        channel->set_revents(events_[i].events);
        activeChannels->push_back(channel); // eventloop拿到了poller返回的所有事件的channel列表
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

最终获取所有被激活且已经修改的Channel，即activeChannels。

EventLoop类

一个EventLoop与一个线程绑定，包含Poller与Channle两大模块，EventLoop的主要作用是在运行过程中进行事件循环，调用Poller执行wait，使得激活Channel执行自己的回调事件，同时执行每个EventLoop自己需要执行的回调，比如监听socket等。

class EventLoop : noncopyable{
public:
    using Functor = std::function;
    EventLoop();
    ~EventLoop();
    // 开启事件循环
    void loop();
    // 退出事件循环
    void quit();
    // 在当前loop中执行cb
    void runInLoop(Functor cb);
    // 把cb放入队列中，唤醒loop所在线程，执行cb
    void queueInLoop(Functor cb);
    // 唤醒loop所在线程
    void wakeup();
    // EventLoop方法 =》 Poller方法
    void updateChannel(Channel* channel);
    void removeChannel(Channel* channel);
    bool hasChannel(Channel* channel);
    // 判断EventLoop对象是否在自己线程里面
    bool isInLoopThread() const { return threadId_ == CurrentThread::tid(); }
private:
    // 通过接收wakeupFd消息来唤醒自己
    void handleRead(); // wake up
    void doPendingFunctors(); // 执行回调

    using ChannelList = std::vector;

    std::atomic_bool looping_; // 原子操作  判断loop是否在事件循环
    std::atomic_bool quit_;  // 标识退出loop循环

    const pid_t threadId_;  // loop所在线程id
    
    Timestamp pollReturnTime_;  // poller返回发生事件的channels的时间点
    std::unique_ptr poller_;  // loop所使用的poller

    int wakeupFd_;  // eventfd 用于线程间通信，通过轮询算法唤醒subloop
    std::unique_ptr wakeupChannel_; // 将eventfd封装成channel

    ChannelList activeChannels_;  // poller监听返回的活跃channel集合

    std::atomic_bool callingPendingFunctors_;  // 判断loop是否有新的回调需要执行
    std::vector pendingFunctors_; // 待处理的回调集合
    std::mutex mutex_; // 互斥锁
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

EventlLoop通过loop方法开启循环，通过quit退出循环。

void EventLoop::loop() {
    looping_ = true;
    quit_ = false;
    while(!quit_) {
        activeChannels_.clear();
        pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
        for(Channel *channel : activeChannels_) {
            // Poller监听哪些channel发生事件了，然后上报给EventLoop，通知channel处理相应的事件
            channel->HandleEvent(pollReturnTime_);
        }
        // 执行当前EventLoop事件循环需要处理的回调操作
        doPendingFunctors();
    }
    LOG_INFO("EventLoop %p stop looping. \n", this);
    looping_ = false;
}

void EventLoop::quit() {
    quit_ = true;
    // 判断执行quit的线程是否是该EventLoop对应的线程，
    // 如果不是，唤醒对应线程
    if(!isInLoopThread()) {
        wakeup();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

不仅如此，loop本身也需要执行对应的回调，如监听socket操作等，通过runInLoop方法设置。在执行操作中，需要判断执行线程是否是EventLoop对应线程，如果是，直接执行，如果不是，将其加入执行队列，并唤醒。

void EventLoop::runInLoop(Functor cb) {
    if(isInLoopThread()) {
        cb();
    }
    else {
        queueInLoop(cb);
    }
}

void EventLoop::queueInLoop(Functor cb) {
    {
        std::unique_lock lock(mutex_);
        pendingFunctors_.emplace_back(cb);
    }

    if(!isInLoopThread() || callingPendingFunctors_) {
        wakeup();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

唤醒方法与接收唤醒方法通过wakeup与handleRead方法实现。每个EventLoop都有一个只用来唤醒操作的eventfd，称为wakeupFd_。当需要唤醒对应EventLoop时，只需要向对应的fd发送1字节数据；而当wakefd接收到1字节数据时，证明该醒来工作了。

void EventLoop::wakeup() {
    uint64_t one = 1;
    ssize_t n = write(wakeupFd_, &one, sizeof one);
    if(n != sizeof one) {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8", n);
    }
}


void EventLoop::handleRead() {
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_, &one, sizeof one);
    if(n != sizeof one) {
        LOG_ERROR("EventLoop::handleRead() reads %lu bytes instead of 8", n);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

EventLoopThread类

muduo的核心就是one loop per thread，所以使用EventLoopThread类将线程与loop绑定起来，启动thread的同时启动对应的loop。

class EventLoopThread : noncopyable {
public:
    using ThreadInitCallBack = std::function;

    EventLoopThread(const ThreadInitCallBack &cb = ThreadInitCallBack(), const std::string &name = std::string());
    ~EventLoopThread();
	// 开启线程，进入事件循环
    EventLoop* startLoop();
private:
    void threadFunc();

    EventLoop *loop_;  // 对应loop
    bool exiting_; // 线程是否结束
    Thread thread_;  // 线程
    // 互斥锁  判断线程与loop是否都创建好绑定完全
    std::mutex mutex_; 
    std::condition_variable cond_; // 条件变量
    ThreadInitCallBack callback_; //线程初始化时的回调
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

所以在stratLoop方法中，当线程启动的时候， 必须确保线程对应的loop已经创建，否则进入阻塞状态等待loop创建成功。

EventLoop* EventLoopThread::startLoop() {

    thread_.start();

    EventLoop *loop = nullptr;
    {
        std::unique_lock lock(mutex_);
        while (loop_ == nullptr)
        {	
            // 当loop不存在时，等待条件变量通知
            cond_.wait(lock);
        }
        loop = loop_;
    }
    return loop;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

EventLoopThreadPool类

在多线程环境下，频繁创建销毁线程是一件费时费力的事情，所以为了提高效率，会创建线程池提前存放一些可用的线程。当需要用到线程时，从线程池的拿出线程使用，而不需要再创建线程；使用完线程后，将线程放入线程池中，避免了销毁线程的开销。

class EventLoopThreadPool : noncopyable {
public:
    using ThreadInitCallBack = std::function;

    EventLoopThreadPool(EventLoop *baseLoop, const std::string &name);
    ~EventLoopThreadPool();
	
    // 设置线程池中线程个数
    void setThreadNum(int numThreads) { numThreads_ = numThreads; }
    
    // 初始化线程池，启动线程池
    void start(const ThreadInitCallBack &cb = ThreadInitCallBack());
	
    // 获取线程池中下一个loop
    EventLoop* getNextLoop();
    
    // 返回loop队列
    std::vector getAllLoops();

    bool started() const { return started_; }
    const std::string name() const { return name_; }
    
private:
    // 主loop，最开始用户创建的loop，在多线程环境下，可看做reactor，用来分发线程
    EventLoop *baseLoop_;
    std::string name_;  // 线程池名字
    bool started_;  // 线程池是否启动
    int numThreads_;  //　线程数量
    int next_; // 标记下次获取loop的下标
    // 线程池中的线程与loop
    std::vector> threads_;
    std::vector loops_;
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

调用start方法开启线程池

void EventLoopThreadPool::start(const ThreadInitCallBack &cb) {
    started_ = true;
    
	// 创建用户指定数量的线程
    for(int i = 0; i(t));
         // 底层创建线程，绑定一个新的EventLoop，并返回该loop的地址
        loops_.push_back(t->startLoop());
    }
	// 如果指定线程为0, 则为单线程环境，监听与IO等操作都由baseloop完成。
    if(numThreads_ == 0 && cb) {
        cb(baseLoop_);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

当新连接到来，需要使用新线程时，调用getNextLoop方法获取下一个空线程使用

EventLoop* EventLoopThreadPool::getNextLoop() {
    EventLoop *loop = baseLoop_;
	// next_标志下一个使用的线程下标
    if(!loops_.empty()) {
        loop = loops_[next_];
        ++next_;
        if(next_ >= loops_.size()) {
            next_ = 0;
        }
    }

    return loop;
}
1
2
3
4
5
6
7
8
9
10
11
12
13