Redis异步实现解析

---

1. Redis 异步方式

Redis 提供了两种主要的连接方式：同步连接和异步连接。选择合适的连接方式可以显著影响应用的性能和响应能力。

1.1 同步连接

同步连接是指客户端与 Redis 服务器之间的一种简单的请求-响应模式。当使用同步连接时，客户端发送一个命令给 Redis 服务器，并在收到服务器响应之后才能继续执行后续的操作。在同步连接中，客户端在执行命令时会阻塞当前线程，等待服务器的响应，直到响应返回或超时。

优点

• 实现简单：代码逻辑直观，易于理解和维护。
• 业务逻辑连贯：操作按顺序执行，业务逻辑没有割裂。

缺点

• 阻塞当前线程：每个命令执行都需要等待响应，可能导致线程阻塞，影响性能。
• 效率较低：在高并发场景下，频繁的阻塞和唤醒线程会增加系统开销。

示例：访问 Redis，并对 counter 实现自增1000次，统计用时

以下是一个使用 hiredis 库的 C 程序示例，该程序连接到 Redis 服务器，对 counter 键进行 1000 次自增操作，并统计所用时间。

编译命令：

gcc redis-test-sync.c -o redis-test-sync -lhiredis

代码示例 (redis-test-sync.c)：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <hiredis/hiredis.h>// 获取当前时间的毫秒数
int current_tick() {int t = 0;struct timespec ti;clock_gettime(CLOCK_MONOTONIC, &ti);t = (int)ti.tv_sec * 1000;t += ti.tv_nsec / 1000000;return t;
}int main(int argc, char **argv) {redisContext *c;redisReply *reply;const char *hostname = "127.0.0.1";int port = 6379;// 设置连接超时时间为1.5秒struct timeval timeout = { 1, 500000 }; // 1.5 seconds// 连接到Redis服务器c = redisConnectWithTimeout(hostname, port, timeout);if (c == NULL || c->err) {if (c) {printf("Connection error: %s\n", c->errstr);redisFree(c);} else {printf("Connection error: can't allocate redis context\n");}exit(1);}// 获取自增次数，默认为1000int num = (argc > 1) ? atoi(argv[1]) : 1000;int before = current_tick();// 执行自增操作for (int i = 0; i < num; i++) {reply = redisCommand(c, "INCR counter"); // +1操作printf("INCR counter: %lld\n", reply->integer);freeReplyObject(reply);}int used = current_tick() - before;printf("After %d exec redis commands, used %d ms\n", num, used);// 断开连接redisFree(c);return 0;
}

输出示例：

INCR counter: 1
INCR counter: 2
...
INCR counter: 1000
After 1000 exec redis commands, used 150 ms

解释：

1. 连接Redis服务器：使用 redisConnectWithTimeout 函数连接到 Redis 服务器，设置超时时间为 1.5 秒。
2. 执行自增操作：循环执行 INCR counter 命令，次数由命令行参数决定，默认为 1000 次。
3. 统计用时：记录操作前后的时间差，计算总耗时。
4. 输出结果：每次自增操作的结果，以及总耗时。

---

1.2 异步连接

异步连接允许客户端在发送命令后不等待响应，可以继续执行其他操作，适用于高性能和高并发场景。异步连接通常通过事件驱动机制实现，可以充分利用多核 CPU 和网络资源。

优点

• 非阻塞操作：不会阻塞当前线程，提升系统的并发处理能力。
• 更高的效率：适合高并发和低延迟要求的应用场景。
• 灵活的业务逻辑：可以在等待响应的同时处理其他任务。

缺点

• 实现复杂：代码逻辑较为复杂，需要处理事件循环和回调函数。
• 业务逻辑割裂：异步操作可能导致业务逻辑分散，难以维护。
• 依赖事件库：通常需要结合事件驱动库（如 libevent、libuv）使用。

1.2.1 Redis 驱动

实现异步连接需要一个 Redis 驱动，该驱动需要将 Redis 连接融入项目中的 Reactor 模式进行管理。具体步骤如下：

1. 实现Redis驱动：

   • 服务端使用异步连接，需要自己实现 Redis 驱动，即将 Redis 连接与项目中的 Reactor 进行融合管理。

2. 设计Redis适配器：

   • 构建Redis事件对象：包括 hiredis 事件对象和 Reactor 事件对象。
   • 适配事件控制：复用项目中的 Reactor 事件循环。

3. hiredis 的封装规则：

   • Reactor 的实现：所有的 IO 由用户实现。
   • 适配器的实现：hiredis 提供事件操作接口，用户需要适配这些接口。不同的网络库和平台对事件操作接口的实现可能不同。

用户需要适配的 hiredis 事件接口包括：

• addRead：添加读事件
• delRead：删除读事件
• addWrite：添加写事件
• delWrite：删除写事件
• cleanup：事件对象释放
• scheduleTimer

1.2.2 示例

以下示例展示了如何实现 Redis 异步连接，包括 Reactor 的实现、Redis 适配器以及主体代码。

第1步：实现 Reactor

文件：reactor.h

#ifndef _MARK_REACTOR_
#define _MARK_REACTOR_#include <sys/epoll.h>
#include <stdio.h>
#include <unistd.h> // read write
#include <fcntl.h> // fcntl
#include <sys/types.h> // listen
#include <sys/socket.h> // socket
#include <errno.h> // errno
#include <arpa/inet.h> // inet_addr htons
#include <assert.h> // assert
#include <stdlib.h> // malloc
#include <string.h> // memcpy memmove#include "chainbuffer/buffer.h" // 自定义缓冲区实现#define MAX_EVENT_NUM 512       // 每次用户拷贝事件的最大数目
#define MAX_CONN ((1<<16)-1)    // 事件对象的最大数目：65535typedef struct event_s event_t;typedef void (*event_callback_fn)(int fd, int events, void *privdata);
typedef void (*error_callback_fn)(int fd, char * err);// Reactor对象，管理IO的全局变量
typedef struct {int epfd;        // epoll文件描述符int listenfd;    // 监听的文件描述符int stop;        // 停止循环标记event_t *events; // 存储所有事件对象，存储在堆上，记得释放int iter;        // 用于遍历events，获取未被使用的位置struct epoll_event fire[MAX_EVENT_NUM]; // 用户态数组，用于拷贝IO事件到用户态
} reactor_t;// 事件对象，sockitem，保存每个fd对应的IO状态
struct event_s {int fd;         // 事件对应的文件描述符reactor_t *r;   // 指向Reactor全局对象buffer_t in;    // 读缓冲，待读取buffer_t out;   // 写缓冲，待发送event_callback_fn read_fn;  // 读回调函数event_callback_fn write_fn; // 写回调函数error_callback_fn error_fn; // 错误回调函数
};// 函数声明
int event_buffer_read(event_t *e);
int event_buffer_write(event_t *e, void * buf, int sz);reactor_t * create_reactor();void release_reactor(reactor_t * r);event_t * new_event(reactor_t *R, int fd,event_callback_fn rd,event_callback_fn wt,error_callback_fn err);void free_event(event_t *e);int set_nonblock(int fd);int add_event(reactor_t *R, int events, event_t *e);int del_event(reactor_t *R, event_t *e);int enable_event(reactor_t *R, event_t *e, int readable, int writeable);void eventloop_once(reactor_t * r, int timeout);void stop_eventloop(reactor_t * r);void eventloop(reactor_t * r);int create_server(reactor_t *R, short port, event_callback_fn func);int event_buffer_read(event_t *e);int event_buffer_write(event_t *e, void * buf, int sz);#endif

文件：reactor.c

#include "reactor.h"// 创建Reactor对象
reactor_t *
create_reactor() {reactor_t *r = (reactor_t *)malloc(sizeof(*r));r->epfd = epoll_create(1);r->listenfd = 0;r->stop = 0;r->iter = 0;r->events = (event_t*)malloc(sizeof(event_t)*MAX_CONN);memset(r->events, 0, sizeof(event_t)*MAX_CONN);memset(r->fire, 0, sizeof(struct epoll_event) * MAX_EVENT_NUM);return r;
}// 释放Reactor对象
void
release_reactor(reactor_t * r) {free(r->events);close(r->epfd);free(r);
}// 获取Reactor的事件堆event上的空闲事件对象
static event_t *
_get_event_t(reactor_t *r) {r->iter++;while (r->events[r->iter & MAX_CONN].fd > 0) {r->iter++;}return &r->events[r->iter];
}// 创建事件对象
event_t *
new_event(reactor_t *R, int fd,event_callback_fn rd,event_callback_fn wt,error_callback_fn err) {assert(rd != 0 || wt != 0 || err != 0);// 获取空闲事件对象event_t *e = _get_event_t(R);// 初始化事件对象e->r = R;e->fd = fd;buffer_init(&e->in, 1024*16);buffer_init(&e->out, 1024*16);e->read_fn = rd;e->write_fn = wt;e->error_fn = err;return e;
}// 释放事件对象分配的buffer空间
void
free_event(event_t *e) {buffer_free(&e->in);buffer_free(&e->out);
}// 设置非阻塞的fd
int
set_nonblock(int fd) {int flag = fcntl(fd, F_GETFL, 0);return fcntl(fd, F_SETFL, flag | O_NONBLOCK);
}// 添加事件
int
add_event(reactor_t *R, int events, event_t *e) {struct epoll_event ev;ev.events = events;ev.data.ptr = e;if (epoll_ctl(R->epfd, EPOLL_CTL_ADD, e->fd, &ev) == -1) {printf("Add event error, fd = %d\n", e->fd);return 1;}return 0;
}// 删除事件
int
del_event(reactor_t *R, event_t *e) {epoll_ctl(R->epfd, EPOLL_CTL_DEL, e->fd, NULL);free_event(e);return 0;
}// 修改事件（读事件 or 写事件）
int
enable_event(reactor_t *R, event_t *e, int readable, int writeable) {struct epoll_event ev;ev.events = (readable ? EPOLLIN : 0) | (writeable ? EPOLLOUT : 0);ev.data.ptr = e;if (epoll_ctl(R->epfd, EPOLL_CTL_MOD, e->fd, &ev) == -1) {return 1;}return 0;
}// 一次事件循环
void
eventloop_once(reactor_t * r, int timeout) {int n = epoll_wait(r->epfd, r->fire, MAX_EVENT_NUM, timeout);for (int i = 0; i < n; i++) {struct epoll_event *e = &r->fire[i];int mask = e->events;if (e->events & EPOLLERR) mask |= EPOLLIN | EPOLLOUT;if (e->events & EPOLLHUP) mask |= EPOLLIN | EPOLLOUT;event_t *et = (event_t*) e->data.ptr;// 处理读事件if (mask & EPOLLIN) {if (et->read_fn)et->read_fn(et->fd, EPOLLIN, et);}// 处理写事件if (mask & EPOLLOUT) {if (et->write_fn)et->write_fn(et->fd, EPOLLOUT, et);else {uint8_t * buf = buffer_write_atmost(&et->out);event_buffer_write(et, buf, buffer_len(&et->out));}}}
}// 停止事件循环
void
stop_eventloop(reactor_t * r) {r->stop = 1;
}// 事件循环
void
eventloop(reactor_t * r) {while (!r->stop) {eventloop_once(r, -1); // 阻塞等待事件}
}// 创建服务器
int
create_server(reactor_t *R, short port, event_callback_fn func) {int listenfd = socket(AF_INET, SOCK_STREAM, 0);if (listenfd < 0) {printf("Create listenfd error!\n");return -1;}struct sockaddr_in addr;memset(&addr, 0, sizeof(struct sockaddr_in));addr.sin_family = AF_INET;addr.sin_port = htons(port);addr.sin_addr.s_addr = INADDR_ANY;// 设置地址重用，允许快速重启服务器int reuse = 1;if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (void *)&reuse, sizeof(int)) == -1) {printf("Reuse address error: %s\n", strerror(errno));return -1;}if (bind(listenfd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)) < 0) {printf("Bind error: %s\n", strerror(errno));return -1;}if (listen(listenfd, 5) < 0) {printf("Listen error: %s\n", strerror(errno));return -1;}if (set_nonblock(listenfd) < 0) {printf("Set nonblock error: %s\n", strerror(errno));return -1;}R->listenfd = listenfd;event_t *e = new_event(R, listenfd, func, 0, 0);add_event(R, EPOLLIN, e);printf("Listening on port: %d\n", port);return 0;
}// 读取数据
int
event_buffer_read(event_t *e) {int fd = e->fd;int num = 0;while (1) {// TODO: 不要在这里使用固定大小的缓冲区char buf[1024] = {0};int n = read(fd, buf, 1024);if (n == 0) { // 半关闭状态printf("Close connection, fd = %d\n", fd);if (e->error_fn)e->error_fn(fd, "Close socket");del_event(e->r, e);close(fd);return 0;} else if (n < 0) { // 异常if (errno == EINTR) // 被中断，重试continue;if (errno == EWOULDBLOCK)  // 阻塞，因为read buffer为空break;printf("Read error, fd = %d, err = %s\n", fd, strerror(errno));if (e->error_fn)e->error_fn(fd, strerror(errno));del_event(e->r, e);close(fd);return 0;} else { // 正常printf("Received data from client: %s", buf);buffer_add(&e->in, buf, n);}// 多次读取的话，按序接在后面num += n;}return num;
}// 向对端发送数据
static int
_write_socket(event_t *e, void * buf, int sz) {int fd = e->fd;while (1) {int n = write(fd, buf, sz);if (n < 0) {if (errno == EINTR)continue;if (errno == EWOULDBLOCK)break;if (e->error_fn)e->error_fn(fd, strerror(errno));del_event(e->r, e);close(e->fd);}return n;}return 0;
}// 写数据
int
event_buffer_write(event_t *e, void * buf, int sz) {buffer_t *r = &e->out;if (buffer_len(r) == 0) {int n = _write_socket(e, buf, sz);if (n == 0 || n < sz) {// 发送失败，将未发送的数据写入缓冲区，并注册写事件buffer_add(&e->out, (char *)buf + n, sz - n);enable_event(e->r, e, 1, 1);return 0;} else if (n < 0) return 0;return 1;}buffer_add(&e->out, (char *)buf, sz);return 1;
}

第2步：实现 Redis 适配器

文件：adapter_async.h

#ifndef _MARK_ADAPTER_
#define _MARK_ADAPTER_#include <hiredis/hiredis.h>
#include <hiredis/async.h>
#include "reactor.h"// Redis事件对象
typedef struct {event_t e;              // Reactor事件对象int mask;               // 存储注册的事件redisAsyncContext *ctx; // hiredis异步上下文
} redis_event_t;// Redis对象读事件回调
static void redisReadHandler(int fd, int events, void *privdata) {(void)fd;(void)events;event_t *e = (event_t*)privdata;redis_event_t *re = (redis_event_t *)(char *)e;redisAsyncHandleRead(re->ctx);
}// Redis对象写事件回调
static void redisWriteHandler(int fd, int events, void *privdata) {(void)fd;(void)events;event_t *e = (event_t*)privdata;redis_event_t *re = (redis_event_t *)(char *)e;redisAsyncHandleWrite(re->ctx);
}/*** @brief 更新Reactor管理的事件对象* @param privdata Redis事件对象* @param flag 要设置的epoll事件类型 * @param remove 1 删除该事件，0 添加该事件*/
static void redisEventUpdate(void *privdata, int flag, int remove) {redis_event_t *re = (redis_event_t *)privdata;reactor_t *r = re->e.r;int prevMask = re->mask;int enable = 0;             // Redis事件对象删除该事件if (remove) {if ((re->mask & flag) == 0) {return;}re->mask &= ~flag;enable = 0;} // Redis事件对象添加该事件else {if (re->mask & flag) {return;    }           re->mask |= flag;enable = 1;}// 对Reactor事件对象的处理// 1. 删除该事件if (re->mask == 0) {del_event(r, &re->e);} // 2. 添加该事件（首次加入）else if (prevMask == 0) {add_event(r, re->mask, &re->e);} // 3. 修改该事件else {// 注册读事件if (flag & EPOLLIN) {enable_event(r, &re->e, enable, 0);} // 注册写事件else if (flag & EPOLLOUT) {enable_event(r, &re->e, 0, enable);}}
}// 添加读事件
static void redisAddRead(void *privdata) {redis_event_t *re = (redis_event_t *)privdata;re->e.read_fn = redisReadHandler;redisEventUpdate(privdata, EPOLLIN, 0);
}// 删除读事件
static void redisDelRead(void *privdata) {redis_event_t *re = (redis_event_t *)privdata;re->e.read_fn = 0;redisEventUpdate(privdata, EPOLLIN, 1);
}// 添加写事件
static void redisAddWrite(void *privdata) {redis_event_t *re = (redis_event_t *)privdata;re->e.write_fn = redisWriteHandler;redisEventUpdate(privdata, EPOLLOUT, 0);
}// 删除写事件
static void redisDelWrite(void *privdata) {redis_event_t *re = (redis_event_t *)privdata;re->e.write_fn = 0;redisEventUpdate(privdata, EPOLLOUT, 1);
}// 释放Redis事件对象
static void redisCleanup(void *privdata) {redis_event_t *re = (redis_event_t *)privdata;reactor_t *r = re->e.r;del_event(r, &re->e);free(re);
}// 绑定Redis异步上下文到Reactor
static int redisAttach(reactor_t *r, redisAsyncContext *ac) { redisContext *c = &(ac->c); // 获取Redis同步上下文redis_event_t *re;          // Redis事件对象/* 确保没有重复绑定 */if (ac->ev.data != NULL)return REDIS_ERR;/* 创建容器用于ctx和读写事件 */re = (redis_event_t*)malloc(sizeof(*re));if (re == NULL) {return REDIS_ERR;}  // 初始化Redis事件对象re->ctx = ac;       // 绑定Redis异步上下文re->e.fd = c->fd;   // 绑定Redis的fdre->e.r = r;        // 绑定Reactorre->mask = 0;       // 初始化事件掩码// 设置hiredis的事件接口，用户实现这些接口ac->ev.addRead = redisAddRead;ac->ev.delRead = redisDelRead;ac->ev.addWrite = redisAddWrite;ac->ev.delWrite = redisDelWrite;ac->ev.cleanup = redisCleanup;ac->ev.data = re;return REDIS_OK;
}#endif

第3步：实现主体代码

编译命令：

gcc chainbuffer/buffer.c redis-test-async.c reactor.c -o redis-test-async -lhiredis

代码示例 (redis-test-async.c)：

#include <hiredis/hiredis.h>
#include <hiredis/async.h>
#include <time.h>#include "reactor.h"
#include "adapter_async.h"static reactor_t *R;
static int cnt, before, num;// 获取当前时间的毫秒数
int current_tick() {int t = 0;struct timespec ti;clock_gettime(CLOCK_MONOTONIC, &ti);t = (int)ti.tv_sec * 1000;t += ti.tv_nsec / 1000000;return t;
}// 回调函数，用于处理Redis响应
void getCallback(redisAsyncContext *c, void *r, void *privdata) {redisReply *reply = r;if (reply == NULL) return;printf("argv[%s]: %lld\n", (char*)privdata, reply->integer);/* 当收到所有回复后，断开连接并停止事件循环 */cnt++;if (cnt == num) {int used = current_tick() - before;printf("After %d exec redis commands, used %d ms\n", num, used);redisAsyncDisconnect(c);}
}// 连接回调函数
void connectCallback(const redisAsyncContext *c, int status) {if (status != REDIS_OK) {printf("Error: %s\n", c->errstr);stop_eventloop(R);return;}printf("Connected...\n");
}// 断开连接回调函数
void disconnectCallback(const redisAsyncContext *c, int status) {if (status != REDIS_OK) {printf("Error: %s\n", c->errstr);stop_eventloop(R);return;}printf("Disconnected...\n");stop_eventloop(R);
}int main(int argc, char **argv) {// 创建Redis异步上下文redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);if (c->err) {/* 暂时让c泄漏 */printf("Error: %s\n", c->errstr);return 1;}// 创建Reactor对象R = create_reactor();// 绑定Reactor对象和Redis异步上下文redisAttach(R, c);// 设置连接和断开回调函数redisAsyncSetConnectCallback(c, connectCallback);redisAsyncSetDisconnectCallback(c, disconnectCallback);// 记录开始时间before = current_tick();// 获取自增次数，默认为1000num = (argc > 1) ? atoi(argv[1]) : 1000;// 发送异步自增命令for (int i = 0; i < num; i++) {redisAsyncCommand(c, getCallback, "count", "INCR counter");}// 运行事件循环eventloop(R);// 释放Reactor对象release_reactor(R);return 0;
}

解释：

1. 创建Redis异步上下文：使用 redisAsyncConnect 函数连接到 Redis 服务器。
2. 创建并绑定Reactor：创建一个 Reactor 对象，并将其与 Redis 异步上下文绑定，以便通过 Reactor 管理 Redis 的事件。
3. 设置回调函数：
   • connectCallback：处理连接成功或失败的回调。
   • disconnectCallback：处理断开连接的回调。
   • getCallback：处理每次 INCR counter 命令的响应，并在完成所有命令后断开连接。
4. 发送异步命令：循环发送 INCR counter 命令，不会阻塞主线程。
5. 运行事件循环：通过 Reactor 的 eventloop 函数处理所有异步事件。
6. 释放资源：事件循环结束后，释放 Reactor 对象。

输出示例：

Connected...
argv[count]: 1
argv[count]: 2
...
argv[count]: 1000
After 1000 exec redis commands, used 150 ms
Disconnected...

参考

0voice · GitHub