socket连接池SocketPool分析(十):libevent

文章目录

1. 1. 用到的libevent的5个接口
   1. 1.1. struct event_base* event_base_new()
   2. 1.2. struct event* event_new(struct event_base* base, evutil_socket_t listenfd, EV_READ, event_callback_fn callback_func, (void*)base)
   3. 1.3. int event_add(struct event* listen_event, const struct timeval *tv)
   4. 1.4. event_base_dispatch(struct event_base* base)
   5. 1.5. event_base_free(struct event_base* base)
2. 2. bufferevent的接口
   1. 2.1. struct bufferevent * bufferevent_socket_new(struct event_base *base, evutil_socket_t fd, int options)
   2. 2.2. void bufferevent_setcb(struct bufferevent *bufev, bufferevent_data_cb readcb, bufferevent_data_cb writecb, bufferevent_event_cb eventcb, void *cbarg)
   3. 2.3. int bufferevent_enable(struct bufferevent *bufev, short event)
   4. 2.4. size_t bufferevent_read(struct bufferevent *bufev, void *data, size_t size)
   5. 2.5. int bufferevent_write(struct bufferevent *bufev,const void *data, size_t size)
   6. 2.6. bufferevent_free(bev);
3. 3. 请看我的gtest_server，是使用libevent实现的：
4. 4. 延伸阅读：
5. 5. 《socket连接池SocketPool分析》系列文章总结：

我还在读大学的时候，C++的库听过最有名的是boost，其次就是这个libevent了，但是一直没有去了解，正好，借着复习UNP的时候学习一下这个大名鼎鼎的网络库。

libevent的官方文档：http://libevent.org

libevent是一个典型的reactor模型。有关reactor模型就请参见文章《socket连接池SocketPool分析(八): 并发服务器》。在这篇文章当中我只谈使用libevent。

用到的libevent的5个接口

• event_base_new()
• event_new()
• event_add()
• event_base_dispatch()
• event_base_free()

使用这些接口需要包含event2/event.h头文件。

struct event_base* event_base_new()

创建一个event_base的指针。类似于epoll当中的创建一个句柄。

int efd = epoll_create(FDSIZE);

以后所有的接口都需要调用这个event_base类型的指针，并且每个线程有且只有一个event_base指针。

struct event* event_new(struct event_base* base, evutil_socket_t listenfd, EV_READ, event_callback_fn callback_func, (void*)base)

创建一个事件event。相当于epoll_ctl当中的EPOLL_CTL_ADD。
第一个参数是根据event_base_new()创建的event_base指针，第二个参数是socket描述符。
第三个是属性：

• EV_TIMEOUT: 超时
• EV_READ: 只要网络缓冲中还有数据，回调函数就会被触发
• EV_WRITE: 只要塞给网络缓冲的数据被写完，回调函数就会被触发
• EV_SIGNAL: POSIX信号量，参考manual吧
• EV_PERSIST: 不指定这个属性的话，回调函数被触发后事件会被删除
• EV_ET: Edge-Trigger边缘触发，参考EPOLL_ET
  这些属性使用|连接起来。

接下来就是回调函数callback_func，它的类型event_callback_fn是这样子的：

typedef void(* event_callback_fn)(evutil_socket_t sockfd, short event_type, void *arg)

最后的参数(void*)base是传递给回调函数callback_func的参数，就是传给了上面的那个void *arg参数。

在我们使用epoll当中，使用epoll_wait()等待，等到有事件触发之后，我们还需要使用if判断是哪一个文件描述符触发的事件。这里直接使用回调函数来做，省去了if判断的功夫。

int event_add(struct event* listen_event, const struct timeval *tv)

相当于epoll_ctl当中的EPOLL_CTL_ADD。

这个函数就是把事件注册到event_base指针上去。如果该事件已经在活动事件队列或者超时队列中，先从中删除。

注意的是这里没有像EPOLL_CTL_DEL这样的删除事件的方法。因为可以使用EV_PERSIST这个属性：不指定这个属性的话，回调函数被触发后事件会被删除。
所以这里不需要手动删除，而是自动删除。

第一个参数就是通过event_new创建的事件，第二个参数则是设定超时值。我一般直接设为NULL，表示永不超时。

event_base_dispatch(struct event_base* base)

启动事件循环，这样才能开始处理发生的事件。循环将一直持续，直到不再有需要关注的事件，或者是遇到event_loopbreak()/event_loopexit()函数。

event_base_free(struct event_base* base)

释放event_base指针，相当于epoll当中的close(efd)。

bufferevent的接口

libevent2提供了bufferevent，直接使用这个就可以不用再用read()函数和write()函数了，而是使用bufferevent_read()和bufferevent_write()来代替。

struct bufferevent * bufferevent_socket_new(struct event_base *base, evutil_socket_t fd, int options)

创建一个bufferevent类型的指针,接下来的bufferevent接口全部都要用到这个指针。

void bufferevent_setcb(struct bufferevent *bufev, bufferevent_data_cb readcb, bufferevent_data_cb writecb, bufferevent_event_cb eventcb, void *cbarg)

在bufferevent指针上绑定读的回调函数，写的回调函数，和异常的回调函数。

int bufferevent_enable(struct bufferevent *bufev, short event)

启动bufferevet，就相当于event_base_dispatch。

size_t bufferevent_read(struct bufferevent *bufev, void *data, size_t size)

调用bufferevent_read()从输入缓存里删除size字节，存到内存的data位置，返回被删除的字节数。
与read()函数相比，read()函数第一个参数是文件描述符，bufferevent_read()函数第一个参数是bufferevent指针。

int bufferevent_write(struct bufferevent *bufev,const void *data, size_t size)

添加数据到输出缓存里，调用bufferevent_write()把内存里data位置的size字节添加到输出缓存的末尾。

bufferevent_free(bev);

使用这个释放资源。

请看我的gtest_server，是使用libevent实现的：

#include <iostream>
#include <unistd.h>
#include <stdlib.h>
#include <assert.h>
#include <thread>
#include <event2/event.h>
#include <event2/bufferevent.h>
#include "SocketPool/server_connection_pool.h"

using std::cout;
using std::cerr;
using std::endl;

const string HOST = "127.0.0.1";
const unsigned PORT = 9999;

const int MAXLINE = 1024;
char buf[MAXLINE];

// 初始化全局的socket连接池
static ServerPoolPtr serverpool_ptr(new ServerPool);

typedef struct mybase {
  SocketObjPtr socketPtr;
  struct event_base* base;
}mybaseStruct;

void do_accept(evutil_socket_t listenfd, short event, void *arg);
void read_cb(struct bufferevent *bev, void *arg);
void write_cb(struct bufferevent *bev, void *arg);
void error_cb(struct bufferevent *bev, short event, void *arg);

int main(int argc, char** argv) {
  // 使用glog来打日志,除错
  google::InitGoogleLogging(argv[0]);
  FLAGS_log_dir = "../log";  
  //从连接池当中取出端口号为9999的连接
  if (!serverpool_ptr->Empty()) {
    SocketObjPtr listener = serverpool_ptr->GetConnection(HOST, PORT);
    //==========================================================
    int listenfd = listener->Get();
    //这里不用epoll了,而是使用封装了epoll的libevent
    //因为这个socket连接池的代码还不需要那么极致的性能
    struct event_base* base = event_base_new();
    //仅仅这一条语句就相当于 int efd = epoll_create(FDSIZE); struct epoll_event ev[EPOLLEVENTS]; 这两条语句
    //可见epoll的语法要比select简洁,而libevent的语法又比epoll简洁
    assert(base != NULL);
    mybaseStruct mbs;
    mbs.socketPtr = listener;
    mbs.base = base;
    struct event* listen_event = event_new(base, listenfd, EV_READ|EV_PERSIST, do_accept, (void*)&mbs);
    //仅仅这一条语句就相当于 struct epoll_event tmp; tmp.events = EPOLLIN; tmp.data.fd = listenfd;这三条语句,其中EV_READ就相当于EPOLLIN,
    //由于在回调函数中需要使用SocketObjPtr,这里我定义一个新的结构体mybase
    event_add(listen_event, NULL);
    //这一条语句就相当于 epoll_ctl(efd, EPOLL_CTL_ADD, listenfd, &tmp); 
  
    //在unp_server当中,这里本来应该定义char buf[MAXLINE];我把它写成全局的
    //这里相比unp_server连while (true)的大循环都省了
    event_base_dispatch(base);
  
    //事件循环结束,清理bufferevent和event
    event_base_free(base);
  
    serverpool_ptr->ReleaseConnection(listener);
  } else {
    cerr << "Sorry, pool is empty!" << endl;
  }
  return 0;
}

//当listenfd上有事件发生时,将调用回调函数do_accept
void do_accept(evutil_socket_t listenfd, short event, void *arg) {
  mybaseStruct* mbsPtr = (mybaseStruct*)arg;
  SocketObjPtr listener = mbsPtr->socketPtr;
  struct event_base* base = mbsPtr->base;
  evutil_socket_t fd = listener->Accept();  
  if (fd < 0) {
    cerr << "accept is error" << endl;
  } 
  //使用bufferevent
  //其实相当于一个高级一点的struct event_base* base = event_base_new();
  struct bufferevent* bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
  //这里设置好当需要读写和错误发生的回调函数
  bufferevent_setcb(bev, read_cb, NULL, error_cb, arg);
  //还记得unp_server当中的删除EPOLLIN事件吗?我当时说的是也可以不删除,只要是read失败了再删除
  bufferevent_enable(bev, EV_READ|EV_WRITE|EV_PERSIST); 
}

void read_cb(struct bufferevent *bev, void *arg) {
  int n;
  evutil_socket_t fd = bufferevent_getfd(bev);
  while (n=bufferevent_read(bev, buf, MAXLINE), n>0) {
    //回显给客户端
    bufferevent_write(bev, buf, n);
    //将buf写回到客户端的时候一定要清空buf,否则的话下次read的时候buf里面会存在客户端上次发送的残留
    memset(buf, 0, sizeof(buf));
  }  
}

void write_cb(struct bufferevent *bev, void *arg) { }

//和unp_server的策略一样,只要是有错误就删除bufferevent事件
void error_cb(struct bufferevent *bev, short event, void *arg) {
  evutil_socket_t fd = bufferevent_getfd(bev);
  if (event & BEV_EVENT_TIMEOUT) {
    LOG(ERROR) << "Time out\n"; 
  } else if (event & BEV_EVENT_EOF) {
    LOG(ERROR) << "Client closed\n";
  }
  bufferevent_free(bev); 
}

延伸阅读：

• 鱼思故渊的libevent源码分析
• sparkliang的libevent源码分析
• libevent中的hello-world.c解讯

《socket连接池SocketPool分析》系列文章总结：

自从我写完《数据库连接池DBPool分析》系列文章到今天已经有2个月的时间了，没想到拖了这么久，但是终于写完了连接池的部分和对网络编程模式的部分，但是一个one loop per thread我并没有实现，因为这个需要进程间通信和线程间通信的知识，我打算再复习一下《UNPv2》这本书，等复习完成之后，我再开始写。