开源项目SMSS开发指南（二）——基于libevent的线程池

libevent是一套轻量级的网络库，基于事件驱动开发。能够实现多线程的多路复用和注册事件响应。本文将介绍libevent的基本功能以及如何利用libevent开发一个线程池。

一. 使用指南

监听服务和注册连接事件

libevent是一个基于事件驱动的网络库，通过在一个事件循环上注册不同的事件以完成线程多路复用。由于libevent采用c语言开发，为了使用方便我们可以将它的功能通过面向对象的设计模式用c++来封装。下面是对常用函数的详细介绍：

（1）event_base_new()：创建（初始化）event_base

event_base代表了一个事件循环上下文，所有需要基于这个事件循环的事件都需要注册在它的上面。如果创建成功，最后需要使用event_base_free()来释放资源。

（2）evconnlistener_new_bind()：绑定一个本地端口并注册网络监听事件

参数说明：

• struct event_base* base 前文创建好的base，事件将关联到这个事件循环上
• evconnlistener_cb cb 事件触发的回调
• void *ptr 回调函数的参数，这个参数可以由用户任意指定，方便在回调函数中使用
• unsigned flags 事件的附加标识，代表事件操作
• int backlog 网络缓存大小
• const struct sockaddr *sa socket地址
• int socklen socket地址长度

函数会返回一个新的evconnlistener，如果创建成功，需要使用evconnlistener_free()来释放资源。

（3）event_base_dispatch()：启动事件循环和事件分发

这个函数会阻塞当前线程，用户可以在事件回调函数中通过event_base_loopbreak()来中断。如果不希望当前线程被堵塞也可以使用event_base_loop()函数。注意，千万不要在回调函数中清理event_base。

代码示例：

// 创建事件循环
ev_base_ = event_base_new();
if (!ev_base_)
{
    return false;
}
sockaddr_in sin;
memset(&sin, 0, sizeof(sin));
sin.sin_family = af_inet;
sin.sin_port = htons(port_);
// 创建socket连接回调
ev_listener_ = evconnlistener_new_bind(
    ev_base_,
    sconnlistenercb,
    this,
    lev_opt_reuseable | lev_opt_close_on_free,
    this->backlog_,
    (sockaddr *)&sin,
    sizeof(sin));
if (!ev_listener_)
{
    return false;
}
while (!quit_)
{
    event_base_loop(ev_base_, evloop_nonblock);
    this_thread::sleep_for(chrono::milliseconds(1));
}
evconnlistener_free(ev_listener_);
event_base_free(ev_base_);

static void sconnlistenercb(struct evconnlistener *listen, evutil_socket_t sock, struct sockaddr *addr, int len, void *ctx)
{
    // 解析客户端ip
    char ip[16] = {0};
    sockaddr_in *addr_in = (sockaddr_in *)addr;
    evutil_inet_ntop(af_inet, &addr_in->sin_addr, ip, sizeof(ip));
    stringstream ss;
    ss << ip << ":" << addr_in->sin_port << " 连接完成...";
    log4cplus_info(simplelogger::get()->loggerref(), ss.str());
    smsserver *server = (smsserver *)ctx;
    int s = sock;

    server->connlistener(s);
}

创建连接和注册读、写、事件监听

（1）bufferevent_socket_new()：创建一个带socket缓存的事件

bufferevent表示一个事件缓存，每当有数据需要读取的时候，它会先将数据从内核态取出再通知用户。顺带提一下，libevent对事件的触发支持两种模式：（et）边沿触发和（lt）水平触发。如果你设置了水平触发，但是通过bufferevent来读取消息，无论你是否将消息接收完成，都不会被反复触发回调。因此，使用bufferevent来接收消息的时候，需要特别关注tcp粘包和长包。

（2）bufferevent_setcb()：设置bufferevent的回调函数

参数说明：

• struct bufferevent* bufev 关联对象
• bufferevent_data_cb readcb 读回调 函数原型void (*bufferevent_data_cb)(struct bufferevent *bev, void* ctx)
• bufferevent_data_cb writecb 写回调 函数原型（同上）
• bufferevent_event_cb eventcb 事件回调 函数原型void (*bufferevent_event_cb)(struct bufferevent *bev, short what, void *ctx)
• void *cbarg 回调函数的最后一个参数，由用户指定

读回调顾名思义就是当有数据的时候会触发的函数，可是写回调什么时候触发？有兴趣的朋友可以自己测试一下。特别需要关注事件回调函数。所有可触发的事件包括：bev_event_reading（读事件），bev_event_writing（写事件），bev_event_eof（结束事件），bev_event_error（错误事件），bev_event_timeout（超时事件），bev_event_connected（连接事件）。如果你是在开发服务端bev_event_connected事件不会被触发，因为连接事件是在bufferevent创建前产生的。bev_event_reading || bev_event_timeout可以用来表示读数据超时，通过这个事件可以侦测心跳代表距离上次读数据已经超时。bev_event_writing || bev_event_timeout可以表示写超时，但是这个事件只会在当有数据需要被发送可是超时未发送成功后才会被触发。

此外，当发生超时事件后，相关的读写操作都会被从bufferevent中移除。如果用户希望继续之前的操作，需要重新注册读/写。

（3）bufferevent_set_timeouts()：设置读/写超时

只有在通过这个函数设置了读/写超时后，在事件回调函数中bev_event_timeout才会生效。

代码示例：

bufferevent *buff_ev_ = bufferevent_socket_new(ev_base_, socket_, bev_opt_close_on_free);
if (!buff_ev_)
{
    return false;
}
// 指定参数
bufferevent_setcb(buff_ev_, sreadcb, swritecb, seventcb, this);
bufferevent_enable(buff_ev_, ev_read | ev_write);
timeval tv = {timeout_, 0};
bufferevent_set_timeouts(buff_ev_, &tv, null);
return true;

读写数据

（1）bufferevent_read()：从缓存中接收数据

通常在读回调中使用，通过返回值判断缓存中是否还有数据

（2）bufferevent_write()：向缓冲写入数据以通过socket发送

返回值表示有多少数据已经被写入进内核

创建基于管道的事件

libevent除了可以用在网络上，还可以和管道（pipe）结合用来生成管道事件。

（1）event_config_new()：创建一个事件配置对象

event_config可以用来创建一个非默认的事件循环，通常使用这个函数配合event_base_new_with_config()创建event_base。最后需要使用event_config_free()来释放资源。

（2）event_new()：创建一个读/写事件

和bufferevent的创建不同，event_new()只会创建一个配套的事件，如果在事件中用户没有对数据进行处理，回调会一直被触发。

代码示例：

// 初始化一对管道，只能在linux系统下使用
int pipefd[2];
if (pipe(pipefd))
{
    return false;
}
// pipefd[0]读取管道 pipefd[1]发送管道
this->pipe_endpoint_ = pipefd[1];
// 创建管道事件
event_config *ev_conf = event_config_new();
event_config_set_flag(ev_conf, event_base_flag_nolock);
this->ev_base_ = event_base_new_with_config(ev_conf);
event_config_free(ev_conf);
if (!ev_base_)
{
    return false;
}

pipe_ev_ = event_new(this->ev_base_, pipefd[0], ev_read | ev_persist, seventcb, this);
event_add(pipe_ev_, 0);

 二、实现线程池

线程池实现原理

libevent可以实现对线程的多路复用，因此我们可以在一个线程中完成服务端对多个客户端的同时读写操作。这样做虽然能够最大限度的利用系统资源，可是无法充分发挥cpu多线程的处理能力。开发高可用和适合高负载的服务端我们依然应该考虑启动多个线程来处理数据。关键是我们如何将事件分发到不同的线程中以保持多个线程的负载均衡。

1. 当服务启动的时候首先创建n条线程。每一个线程对应一个事件循环event_base。
2. 主线程负责监听指定端口并在连接的回调函数中处理新连接套接字的处理。
3. 当有新的客户端连接后，主线程会把套接字先保存在一个队列中。扫描当前所有线程的处理量，选择负载最小的线程利用管道发送一个信号（‘c’）。对应线程的事件循环在管道的读事件中从队列中获取这个套接字，并建立对应的bufferevent进行处理。当前线程负载量+1。
4. 客户端断开后通知bufferevent所在的线程将负载量减一。

smss源码阅读

相关的源码文件为sms_server，work_group，work_thread和socket_manager

服务初始化，注册连接监听事件并初始化线程组

bool smsserver::init()
{
    // 创建事件循环
    ev_base_ = event_base_new();
    if (!ev_base_)
    {
        return false;
    }
    sockaddr_in sin;
    memset(&sin, 0, sizeof(sin));
    sin.sin_family = af_inet;
    sin.sin_port = htons(port_);
    // 创建socket连接回调
    ev_listener_ = evconnlistener_new_bind(
        ev_base_,
        sconnlistenercb,
        this,
        lev_opt_reuseable | lev_opt_close_on_free,
        this->backlog_,
        (sockaddr *)&sin,
        sizeof(sin));
    if (!ev_listener_)
    {
        return false;
    }
    // 创建线程组管理类
    boss_ = new workgroup(thread_num_);
    boss_->init();
    return true;
}

线程组负责管理所有的线程

bool workgroup::init()
{
    // 直接初始化指定的工作线程
    for (int i = 0; i < num_; i++)
    {
        int id = group_.size() + 1;
        workthread *work = new workthread(this, id, net_bus_);
        if (!work->init())
        {
            return false;
        }
        work->start(); // thread start...
        group_.push_back(work);
        // 将当前初始化完成的工作线程注册进消息总线
        net_bus_->regist(work); // regist thread to netbus
    }
    return true;
}

每一个线程在初始化的时候会创建一条管道并在自己的事件循环上注册对应的读回调，对外部暴露notify方法用来激活事件

bool workthread::init()
{
    // 初始化一对管道，只能在linux系统下使用
    int pipefd[2];
    if (pipe(pipefd))
    {
        return false;
    }
    // pipefd[0]读取管道 pipefd[1]发送管道
    this->pipe_endpoint_ = pipefd[1];
    // 创建管道事件
    event_config *ev_conf = event_config_new();
    event_config_set_flag(ev_conf, event_base_flag_nolock);
    this->ev_base_ = event_base_new_with_config(ev_conf);
    event_config_free(ev_conf);
    if (!ev_base_)
    {
        return false;
    }

    pipe_ev_ = event_new(this->ev_base_, pipefd[0], ev_read | ev_persist, seventcb, this);
    event_add(pipe_ev_, 0);
    return true;
}

void workthread::notify(const char *sign)
{
    // 激活
    int re = write(this->pipe_endpoint_, sign, 1);
    if (re <= 0)
    {
        log4cplus_error(simplelogger::get()->loggerref(), "管道激活失败");
    }
}

在读回调中获取套接字，创建连接管理对象socketmanager

void workthread::activated(int fd)
{
    char buf[2] = {0};
    int re = read(fd, buf, 1);
    socket_list_mtx_.lock();
    if (strcmp(buf, "c") == 0) // 通知有新的客户端连接
    {
        // new client connect, create socketmanager
        if (socket_list_.empty())
        {
            socket_list_mtx_.unlock();
            return;
        }
        // 读取一条套接字
        int client_sock = socket_list_.front();
        socket_list_.pop_front();
        // 创建socketmanager
        socketmanager *manager = new socketmanager(this, ev_base_, client_sock, appcontext::get()->client_timeout());
        manager->init();
        sock_manager_list_.push_back(manager);
        stringstream ss;
        ss << "socketmanager:" << client_sock << " 创建完成";
        log4cplus_debug(simplelogger::get()->loggerref(), ss.str());
    }

    socket_list_mtx_.unlock();
}

客户端连接后将创建的套接字交给负载最小的线程处理

void workgroup::createconnection(int sock)
{
    int min = -1;
    workthread *work = nullptr;
    // 遍历寻找负载最轻的线程
    for (auto it = group_.begin(); it != group_.end(); it++)
    {
        if (min == -1)
        {
            min = (*it)->connect_num();
            work = (*it);
        }
        else if ((*it)->connect_num() < min)
        {
            min = (*it)->connect_num();
            work = (*it);
        }
    }
    // 添加一条socket fd进队列并通过管道激活
    work->addsocket(sock);
    work->notify("c");
}

 

完整源码已经发布在上。

相关文章：《开源项目smss开发指南》