開源專案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多執行緒的處理能力，開發高可用和適合高負載的服務端我們依然應該考慮啟動多個執行緒來處理資料，關鍵是我們如何將事件分發到不同的執行緒中以保持多個執行緒的負載均衡，

當服務啟動的時候首先創建N條執行緒，每一個執行緒對應一個事件回圈event_base，
主執行緒負責監聽指定埠并在連接的回呼函式中處理新連接套接字的處理，
當有新的客戶端連接后，主執行緒會把套接字先保存在一個佇列中，掃描當前所有執行緒的處理量，選擇負載最小的執行緒利用管道發送一個信號（‘c’），對應執行緒的事件回圈在管道的讀事件中從佇列中獲取這個套接字，并建立對應的bufferevent進行處理，當前執行緒負載量+1，
客戶端斷開后通知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開發指南》

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/73337.html

標籤：C++

上一篇：oi筆記——抽象的深度優先搜索

下一篇：條款01：視C++為一個語言聯邦