📊 bufferevent详解#

一、概述#

在libevent中，bufferevent是一个高级抽象，它封装了底层的I/O事件和缓冲区管理，使得开发者可以更方便地处理网络I/O操作。bufferevent提供了一种机制，允许应用程序在数据可读或可写时得到通知，而不需要手动处理底层的I/O事件。

bufferevent内部有两个缓冲区：一个用于读取数据，一个用于写入数据。当底层socket有数据可读时，bufferevent会自动将数据读入读缓冲区，并触发读回调；当写缓冲区有数据时，bufferevent会自动将数据写入底层socket，并触发写回调。

二、bufferevent工作流程#

2.1 bufferevent架构#

1
┌─────────────────────────────────────────────────────────┐
2
│                    bufferevent                       │
3
├─────────────────────────────────────────────────────────┤
4
│  ┌────────────┐  ┌────────────┐  ┌────────────┐      │
5
│  │  输入缓冲区  │  │  输出缓冲区  │  │  事件回调  │      │
6
│  └─────┬──────┘  └─────┬──────┘  └─────┬──────┘      │
7
│        │               │               │              │
8
├────────┼───────────────┼───────────────┼──────────────┤
9
│        │               │               │              │
10
├────────▼───────────────▼───────────────▼──────────────┤
11
│                 底层event机制                         │
12
└─────────────────────────────────────────────────────────┘

2.2 bufferevent数据流程#

1
┌─────────────────────────────────────────────────────────┐
2
│                bufferevent数据流程                   │
3
├─────────────────────────────────────────────────────────┤
4
│ 1. 网络数据到达                                       │
5
│ 2. 底层event触发读事件                                 │
6
│ 3. bufferevent自动读取数据到输入缓冲区                    │
7
│ 4. 当数据达到读水位时，触发用户读回调                      │
8
│ 5. 用户通过bufferevent_read读取数据                    │
9
│ 6. 用户通过bufferevent_write写入数据                   │
10
│ 7. 数据被写入输出缓冲区                                  │
11
│ 8. 当输出缓冲区有数据且可写时，自动写入网络                  │
12
│ 9. 当输出缓冲区数据低于写水位时，触发用户写回调                │
13
└─────────────────────────────────────────────────────────┘

三、bufferevent结构体#

1
struct bufferevent {
2
    struct event_base *be_base;
3
    const struct bufferevent_ops *be_ops;
4
    struct event ev_read;       // 读事件
5
    struct event ev_write;      // 写事件
6
    struct evbuffer *input;     // 读缓冲区
7
    struct evbuffer *output;    // 写缓冲区
8

9
    bufferevent_data_cb readcb;   // 读回调
10
    bufferevent_data_cb writecb;  // 写回调
11
    bufferevent_event_cb errorcb; // 错误回调
12
    void *cbarg;
13

14
    struct timeval timeout_read;  // 读超时
15
    struct timeval timeout_write; // 写超时
16

17
    short enabled;
18
    short options;
19
};

四、bufferevent类型#

类型	说明
`bufferevent_socket`	基于socket的bufferevent
`bufferevent_async`	Windows IOCP
`bufferevent_pair`	用于线程间通信
`bufferevent_filter`	数据过滤
`bufferevent_openssl`	SSL/TLS加密

四、创建与销毁#

4.1 创建bufferevent#

1
/* 创建基于socket的bufferevent
2
 * @param base   event_base
3
 * @param fd     文件描述符，如果为-1，后续需要调用bufferevent_socket_connect
4
 * @param options 选项标志
5
 */
6
struct bufferevent *bufferevent_socket_new(struct event_base *base,
7
    evutil_socket_t fd, int options);
8

9
/* 释放bufferevent */
10
void bufferevent_free(struct bufferevent *bev);

4.2 选项标志#

1
#define BEV_OPT_CLOSE_ON_FREE    0x01  // 释放时关闭底层文件描述符
2
#define BEV_OPT_THREADSAFE       0x02  // 使bufferevent线程安全
3
#define BEV_OPT_DEFER_CALLBACKS  0x04  // 延迟回调执行
4
#define BEV_OPT_UNLOCK_CALLBACKS 0x08  // 回调时不持有锁

五、回调函数#

5.1 回调函数类型#

1
/* 读/写回调函数类型
2
 * @param bev   bufferevent指针
3
 * @param ctx   用户提供的参数
4
 */
5
typedef void (*bufferevent_data_cb)(struct bufferevent *bev, void *ctx);
6

7
/* 事件回调函数类型
8
 * @param bev   bufferevent指针
9
 * @param what  事件标志
10
 * @param ctx   用户提供的参数
11
 */
12
typedef void (*bufferevent_event_cb)(struct bufferevent *bev, short what, void *ctx);

5.2 事件标志#

1
#define BEV_EVENT_READING    0x01  // 读操作时发生
2
#define BEV_EVENT_WRITING    0x02  // 写操作时发生
3
#define BEV_EVENT_EOF        0x10  // 遇到文件结束符
4
#define BEV_EVENT_ERROR      0x20  // 发生错误
5
#define BEV_EVENT_TIMEOUT    0x40  // 超时
6
#define BEV_EVENT_CONNECTED  0x80  // 连接成功

5.3 设置回调#

1
/* 设置回调函数
2
 * @param readcb  读回调，当读缓冲区有数据时触发，NULL表示禁用
3
 * @param writecb 写回调，当写缓冲区数据发送完毕时触发，NULL表示禁用
4
 * @param eventcb 事件回调，连接、错误等事件
5
 * @param cbarg   回调函数的参数
6
 */
7
void bufferevent_setcb(struct bufferevent *bufev,
8
    bufferevent_data_cb readcb, bufferevent_data_cb writecb,
9
    bufferevent_event_cb eventcb, void *cbarg);

六、启用和禁用事件#

1
/* 启用事件
2
 * @param event EV_READ, EV_WRITE, 或两者的组合
3
 */
4
int bufferevent_enable(struct bufferevent *bufev, short event);
5

6
/* 禁用事件 */
7
int bufferevent_disable(struct bufferevent *bufev, short event);
8

9
/* 获取当前启用的事件 */
10
short bufferevent_get_enabled(struct bufferevent *bufev);

七、缓冲区操作#

7.1 获取缓冲区#

1
/* 获取输入缓冲区（读缓冲区） */
2
struct evbuffer *bufferevent_get_input(struct bufferevent *bufev);
3

4
/* 获取输出缓冲区（写缓冲区） */
5
struct evbuffer *bufferevent_get_output(struct bufferevent *bufev);

7.2 读写数据#

1
/* 向输出缓冲区写入数据 */
2
int bufferevent_write(struct bufferevent *bufev, const void *data, size_t size);
3

4
/* 向输出缓冲区写入格式化数据 */
5
int bufferevent_write_buffer(struct bufferevent *bufev, struct evbuffer *buf);
6

7
/* 从输入缓冲区读取数据 */
8
size_t bufferevent_read(struct bufferevent *bufev, void *data, size_t size);
9

10
/* 从输入缓冲区读取数据到evbuffer */
11
int bufferevent_read_buffer(struct bufferevent *bufev, struct evbuffer *buf);

八、水位控制#

水位（Watermarks）用于控制缓冲区的大小，当缓冲区中的数据量达到水位时，会触发相应的行为。

8.1 读水位#

水位类型	说明
低水位	读缓冲区数据量达到此值时，才会触发读回调。默认为0，表示有数据就触发
高水位	读缓冲区数据量达到此值时，停止读取数据，直到缓冲区数据量降低

8.2 写水位#

水位类型	说明
低水位	写缓冲区数据量低于此值时，才会触发写回调。默认为0

8.3 设置水位#

1
/* 设置水位
2
 * @param events    EV_READ 或 EV_WRITE
3
 * @param lowmark   低水位
4
 * @param highmark  高水位，0表示无限制
5
 */
6
void bufferevent_setwatermark(struct bufferevent *bufev, short events,
7
    size_t lowmark, size_t highmark);

8.4 水位示例#

1
// 设置读高水位为1MB，防止内存耗尽
2
bufferevent_setwatermark(bev, EV_READ, 0, 1024*1024);
3

4
// 设置读低水位为100字节，只有数据量>=100时才触发读回调
5
bufferevent_setwatermark(bev, EV_READ, 100, 0);

九、超时设置#

1
/* 设置超时
2
 * @param timeout_read  读超时，NULL表示禁用
3
 * @param timeout_write 写超时，NULL表示禁用
4
 */
5
int bufferevent_set_timeouts(struct bufferevent *bufev,
6
    const struct timeval *timeout_read, const struct timeval *timeout_write);

十、连接操作#

10.1 服务端连接#

1
/* 获取底层socket */
2
evutil_socket_t bufferevent_getfd(struct bufferevent *bufev);
3

4
/* 获取所属的event_base */
5
struct event_base *bufferevent_get_base(struct bufferevent *bufev);

10.2 客户端连接#

1
/* 连接服务器
2
 * @param addr 服务器地址
3
 * @param socklen 地址长度
4
 */
5
int bufferevent_socket_connect(struct bufferevent *bev,
6
    const struct sockaddr *addr, int socklen);
7

8
/* 通过主机名连接 */
9
int bufferevent_socket_connect_hostname(struct bufferevent *bev,
10
    struct evdns_base *dns_base, int family, const char *hostname, int port);
11

12
/* 获取连接错误码 */
13
int bufferevent_socket_get_dns_error(struct bufferevent *bev);

十一、使用示例#

11.1 TCP服务端#

1
#include <stdio.h>
2
#include <stdlib.h>
3
#include <string.h>
4
#include <event2/buffer.h>
5
#include <event2/bufferevent.h>
6
#include <event2/listener.h>
7
#include <event2/event.h>
8

9
static void read_cb(struct bufferevent *bev, void *ctx) {
10
    struct evbuffer *input = bufferevent_get_input(bev);
11
    struct evbuffer *output = bufferevent_get_output(bev);
12

13
    // 回显数据
14
    evbuffer_add_buffer(output, input);
15
}
16

17
static void event_cb(struct bufferevent *bev, short events, void *ctx) {
18
    if (events & BEV_EVENT_ERROR) {
19
        perror("Error from bufferevent");
20
    }
21
    if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
22
        bufferevent_free(bev);
23
    }
24
}
25

26
static void accept_conn_cb(struct evconnlistener *listener,
27
    evutil_socket_t fd, struct sockaddr *address, int socklen, void *ctx) {
28

29
    struct event_base *base = evconnlistener_get_base(listener);
30
    struct bufferevent *bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE);
31

32
    bufferevent_setcb(bev, read_cb, NULL, event_cb, NULL);
33
    bufferevent_enable(bev, EV_READ | EV_WRITE);
34
}
35

36
int main() {
37
    struct event_base *base = event_base_new();
38

39
    struct sockaddr_in sin;
40
    memset(&sin, 0, sizeof(sin));
41
    sin.sin_family = AF_INET;
42
    sin.sin_port = htons(8080);
43
    sin.sin_addr.s_addr = htonl(INADDR_ANY);
44

45
    struct evconnlistener *listener = evconnlistener_new_bind(base,
46
        accept_conn_cb, NULL, LEV_OPT_CLOSE_ON_FREE | LEV_OPT_REUSEABLE,
47
        -1, (struct sockaddr*)&sin, sizeof(sin));
48

49
    event_base_dispatch(base);
50

51
    evconnlistener_free(listener);
52
    event_base_free(base);
53
    return 0;
54
}

11.2 TCP客户端#

1
#include <stdio.h>
2
#include <stdlib.h>
3
#include <string.h>
4
#include <event2/buffer.h>
5
#include <event2/bufferevent.h>
6
#include <event2/event.h>
7

8
static void read_cb(struct bufferevent *bev, void *ctx) {
9
    char buf[1024];
10
    size_t n;
11
    while ((n = bufferevent_read(bev, buf, sizeof(buf))) > 0) {
12
        fwrite(buf, 1, n, stdout);
13
    }
14
}
15

16
static void event_cb(struct bufferevent *bev, short events, void *ctx) {
17
    if (events & BEV_EVENT_CONNECTED) {
18
        printf("Connected!\n");
19
        // 发送数据
20
        bufferevent_write(bev, "Hello, Server!\n", 15);
21
    } else if (events & BEV_EVENT_ERROR) {
22
        printf("Error: %s\n", strerror(errno));
23
    }
24
    if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
25
        bufferevent_free(bev);
26
    }
27
}
28

29
int main() {
30
    struct event_base *base = event_base_new();
31
    struct bufferevent *bev = bufferevent_socket_new(base, -1, BEV_OPT_CLOSE_ON_FREE);
32

33
    bufferevent_setcb(bev, read_cb, NULL, event_cb, NULL);
34
    bufferevent_enable(bev, EV_READ | EV_WRITE);
35

36
    struct sockaddr_in sin;
37
    memset(&sin, 0, sizeof(sin));
38
    sin.sin_family = AF_INET;
39
    sin.sin_port = htons(8080);
40
    inet_pton(AF_INET, "127.0.0.1", &sin.sin_addr);
41

42
    bufferevent_socket_connect(bev, (struct sockaddr*)&sin, sizeof(sin));
43

44
    event_base_dispatch(base);
45
    event_base_free(base);
46
    return 0;
47
}

其他相关资料请移步到个人主页

📤 bufferevent详解 - libevent源码深度解析系列（七）