Redis2——协议与异步方式

Redis2——协议与异步方式

本文讲述了Redis pipeline技术，它被用于一次发送和执行多个命令；事务的ACID特性，Redis只能部分满足；最后介绍了实现了Redis客户端的同步连接和异步连接方式。

1. Redis Pipeline

Redis Pipeline 是一种在 Redis 中批量执行命令的技术，用于减少客户端与 Redis 服务器之间通信的开销，从而提高性能。Redis Pipeline是一种客户端提供的技术。

python代码示例如下：

import redis# 创建 Redis 连接
r = redis.Redis(host='localhost', port=6379, db=0)# 使用 pipeline
pipe = r.pipeline()# 批量命令
pipe.set('foo', 'bar')
pipe.get('foo')
pipe.incr('counter')
pipe.mset({'name': 'Alice', 'age': 30})# 执行管道中的所有命令
responses = pipe.execute()# 打印响应
print(responses)

2. Redis事务

定义：事务是由用户定义的一系列操作，被视为一个整体，要么全部执行，要么都不执行，不可分割，其中间状态不能被访问（原子性）。

2.1 无锁事务控制（乐观事务控制）

Redis 的乐观事务主要基于其 WATCH 和 MULTI/EXEC 机制，旨在实现无锁事务控制。通过监控键值变化避免事务冲突，适合高并发、轻量事务。

Redis对于数据的管理使用的是单线程模式，所有的命令会在同一个线程中执行，这使得redis中的大部分命令天然是原子操作，无序借助CAS实现。

1. 监控键值 (WATCH)
   Redis 使用 WATCH 命令来监视一个或多个键。如果在事务执行之前，任何一个被监视的键发生了变化，事务将被中止。
2. 事务定义 (MULTI)
   事务通过 MULTI 命令开始，所有后续命令被加入事务队列。
3. 执行事务 (EXEC)
   当执行 EXEC 时，Redis 检查被监视的键是否被修改过。如果没有，事务中的命令会依次执行；如果有，事务会被放弃。
4. 回滚机制(DISCARD)
   Redis 的乐观事务没有真正的回滚机制。如果某条命令出错，其他命令依然会执行。

乐观锁的概念

乐观锁是一种基于 “假设冲突较少” 的并发控制策略。它的核心思想是：在操作数据时假设不会发生并发冲突，仅在更新时检查是否有冲突，如果检测到冲突，则采取相应的处理（如重试或报错）。

这种锁机制并不会阻塞其他事务的访问，而是允许多个事务同时操作数据，但在最终提交时检查数据的一致性。

2.2 事务语句与lua脚本

开启事务

MULTI

提交事务

EXEC

回滚事务

DISCARD

监控键值变化

WATCH key

如果事务执行之前，被监控的key值发生了变化，就会导致事务执行失败。

例子如下：

WATCH account1 account2      # 监控两个键
val1 = GET account1          # 获取 account1 的余额
if val1 >= 100:              # 判断余额是否足够MULTI                    # 开始事务DECRBY account1 100      # 从 account1 扣减 100INCRBY account2 100      # 给 account2 增加 100EXEC                     # 提交事务
else:UNWATCH                  # 取消监控# 返回余额不足的错误

Lua脚本实现事务的原子性

执行lua脚本

Redis中内置了一个lua虚拟机。在工程实践中，大多数情况下是使用lua脚本来实现事务的原子性。

EVAL script numkeys key [key ...] [arg ...]

举个例子，将key值加倍并返回：

先定义一个lua脚本

local key = KEYS[1]; 
local val = redis.call("get", key); 
if not val thenval = 1000
end
redis.call("set", key, val * 2); 
return 2 * val;

执行lua脚本

eval 'local key ... val;' 1 val

缓存脚本

但是这样每次都发送完整的lua脚本会造成流量浪费，我们可以先使用script load将脚本发送给redis服务器，得到一个标识符，然后之后就可以发送标识符代替脚本。

发送脚本，获得标识符

script load <script string>

使用标识符执行脚本

evalsha <script sha code> numkeys key [key ...] [arg ...]

管理脚本标识符

在服务器启动时，先清空原有的脚本缓存

script flush

然后将所有脚本发送给redis并使用一个unordered_map进行缓存，从而可以随时取用。

2.3 事务特性ACID

1. 原子性(Atomicity) 事务是一个不可分割的单位，要么全部执行，要么全都不执行，其它事务不能访问其中间状态。

   Redis事务具备原子性，但由于Redis不支持回滚，因此即使事务中某些操作执行失败，整个事务也会继续执行下去，直到执行完毕。

2. 一致性(Consistency)包括数据库本身的完整性约束的一致性和用户定义的逻辑上的一致性。前者举例如类型约束、非空约束R、唯一约束、外键约束等，后者举例如银行转账事务前后，总金额应该保持不变。

   Redis不支持逻辑上的一致性，因为它允许事务中的部分操作执行失败。

3. 隔离性(Isolation)并发事务之间的影响程度。由于Redis是单线程执行命令，因此天然具有隔离性。

4. 持久性(Durability)事务的操作的结果是否会持久化到磁盘中，即使数据库重启或崩溃，事务的数据也可以被恢复。

   Redis是基于内存的数据库，其事务是否具备持久性取决于持久化策略和配置：

   • RDB（Redis Database）持久化：通过快照的方式定期将内存中的数据保存到磁盘。

   • AOF（Append-Only File）持久化：通过将每个写命令追加到日志文件中，来记录所有写操作。

   设置持久化策略为每个写命令都持久化到磁盘:

   # redis.conf
   appendonly yes
   appendfsync always 
   

综上，Redis事务具备原子性和隔离性，不具备一致性，是否具备持久性取决于持久化策略。

3. 通信方式

3.1 hiredis库

如果是C++，可以使用redis自带的hireds库进行连接。下面介绍一下hiredis的主要接口：

// 建立redis同步连接
redisContext *redisConnect(const char *ip, int port);
// 建立redis异步连接
redisContext *redisConnectNonBlock(const char *ip, int port);
// 发送命令给redis，如果是同步连接，成功发送返回值为reply，否则返回NULL
void *redisCommand(redisContext *c, const char *format, ...);
// 发送异步命令给redis，还可以设置返回后的回调函数
int redisAsyncCommand(redisAsyncContext *ac, redisCallbackFn *fn, void *privdata, const char *format, ...);
// 获取reply
int redisGetReply(redisContext *c, void **reply);
// 销毁reply
void freeReplyObject(void *reply);
// 销毁redis连接上下文
void redisFree(redisContext *c);

3.2 同步连接

同步连接是指与redis进行同步通信的连接，发送数据和接收数据时可能需要等待从而造成线程阻塞，大量时间被浪费在等待IO传输上。因此在业务中一般不采用这种方式。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <hiredis/hiredis.h>int main() {redisContext *c;redisReply *reply;struct timeval timeout = {1, 500000}; // 1.5 secondsc = redisConnectWithTimeout("127.0.0.1", 6379, timeout);if (c == NULL || c->err){printf("Connecting fails\n");if (c != NULL){printf("error: %s\n", c->errstr);redisFree(c);}exit(1);}reply = redisCommand(c, "set name eeuu");if (c->err){printf("c->err: %d\n", c->err);perror("redis:");}else{printf("replay type: %d\n", reply->type);printf("%s\n", reply->str);}freeReplyObject(reply);redisFree(c);return 0;
}

3.3 异步连接

Redis异步连接依靠非阻塞IO实现，发送命令时不会等待结果返回，而是设置回调函数，当该命令返回结果时自动调用该回调函数来处理结果。

3.3.1 hiredis管理监听事件接口

hiredis主要通过IO事件监测回调函数和读写数据的接口与底层的网络IO层交互。

下面看一下IO检测接口：

typedef struct redisAsyncContext {.../* Event library data and hooks */struct {void *data;/* Hooks that are called when the library expects to start reading/writing. These functions should be idempotent. */void (*addRead)(void *privdata);	// 注册读事件void (*delRead)(void *privdata);	// 注销读事件void (*addWrite)(void *privdata);	// 注册写事件void (*delWrite)(void *privdata);	// 注销写事件void (*cleanup)(void *privdata);	// 清空事件 void (*scheduleTimer)(void *privdata, struct timeval tv);	// 注册定时事件} ev;
} redisAsyncContext;

我们可以在创建redisAsyncContxt时进行设置。

当监听的事件发生时，需要网络层调用hiredis提供的处理函数。

void redisAsyncHandleRead(redisAsyncContext *ac);
void redisAsyncHandleWrite(redisAsyncContext *ac);
void redisAsyncHandleTimeout(redisAsyncContext *ac);

也就是说，网络层需要提供管理IO事件的函数以及当事件发生时调用读写函数。

3.3.2 hiredis + libevent

libevent 是一个高效、跨平台的事件驱动编程库，主要用于构建高性能的网络应用程序和服务。它抽象化了底层的 I/O 多路复用机制，提供了一套简单的 API，用于处理事件通知、定时器和异步 I/O。

在linux平台上，libevent采用epoll实现IO复用。

适配libevent的IO事件管理函数：

static void redisLibeventAddRead(void *privdata) {redisLibeventUpdate(privdata, EV_READ, 0);
}static void redisLibeventDelRead(void *privdata) {redisLibeventUpdate(privdata, EV_READ, 1);
}static void redisLibeventAddWrite(void *privdata) {redisLibeventUpdate(privdata, EV_WRITE, 0);
}static void redisLibeventDelWrite(void *privdata) {redisLibeventUpdate(privdata, EV_WRITE, 1);
}static void redisLibeventCleanup(void *privdata) {redisLibeventEvents *e = (redisLibeventEvents*)privdata;if (!e) {return;}event_del(e->ev);event_free(e->ev);e->ev = NULL;if (e->state & REDIS_LIBEVENT_ENTERED) {e->state |= REDIS_LIBEVENT_DELETED;} else {redisLibeventDestroy(e);}
}

看一下hiredis自带的适配libevent的事件派发函数：

static void redisLibeventHandler(int fd, short event, void *arg) {((void)fd);redisLibeventEvents *e = (redisLibeventEvents*)arg;e->state |= REDIS_LIBEVENT_ENTERED;#define CHECK_DELETED() if (e->state & REDIS_LIBEVENT_DELETED) {\redisLibeventDestroy(e);\return; \}if ((event & EV_TIMEOUT) && (e->state & REDIS_LIBEVENT_DELETED) == 0) {redisAsyncHandleTimeout(e->context);CHECK_DELETED();}if ((event & EV_READ) && e->context && (e->state & REDIS_LIBEVENT_DELETED) == 0) {redisAsyncHandleRead(e->context);CHECK_DELETED();}if ((event & EV_WRITE) && e->context && (e->state & REDIS_LIBEVENT_DELETED) == 0) {redisAsyncHandleWrite(e->context);CHECK_DELETED();}e->state &= ~REDIS_LIBEVENT_ENTERED;#undef CHECK_DELETED
}

3.3.3 hiredis + 自定义reactor

自定义一个reactor模式的网络层的思路类似libevent，主要是要适配hiredis的接口。

以下是采用异步连接时的主函数，其中有3个hiredis提供的接口：

1. redisAsyncConnect是hiredis的接口，可以异步建立连接，当连接成功建立后会自动调用用户自定义的ConnectCallback。
2. redisAsyncSetConnectCallback设置连接建立后回调函数
3. redisAsyncSetDisconnectCallback设置连接断开后的回调函数

int main(int argc, char **argv) {R = create_reactor();redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);if (c->err) {/* Let *c leak for now... */printf("Error: %s\n", c->errstr);return 1;}redisAttach(R, c);redisAsyncSetConnectCallback(c, connectCallback);redisAsyncSetDisconnectCallback(c, disconnectCallback);eventloop(R);release_reactor(R);return 0;
}

剩下还有3个用户定义的函数：

1. create_reactor创建reactor的上下文环境，包括epfd、events缓冲区等。

2. redisAttach，主要负责将reactor上下文环境与连接上下文绑定，代码如下：

   static int redisAttach(reactor_t *r, redisAsyncContext *ac) {redisContext *c = &(ac->c);redis_event_t *re;/* Nothing should be attached when something is already attached */if (ac->ev.data != NULL)return REDIS_ERR;/* Create container for ctx and r/w events */re = (redis_event_t*)hi_malloc(sizeof(*re));if (re == NULL)return REDIS_ERR;re->ctx = ac;re->e.fd = c->fd;re->e.r = r;// dont use event buffer, using hiredis's bufferre->e.in = NULL;re->e.out = NULL;re->mask = 0;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;
   }
   

   可见其主要是创建并填充了一个redis_event_t，然后为redis异步连接上下文设置回调函数，最后将redis_event_t交给连接上下文保存。

   redis_event_t是对reactor的event_t的扩展，结构体如下：

   typedef struct {event_t e;	// 保存fd、epoll上下文、缓冲区、回调函数int mask;	// 管理已注册的事件redisAsyncContext *ctx;
   } redis_event_t;typedef struct event_s event_t;struct event_s {int fd;reactor_t *r;buffer_t *in;buffer_t *out;event_callback_fn read_fn;event_callback_fn write_fn;error_callback_fn error_fn;
   };
   

   为异步连接上下文设置回调函数是关键步骤，分别对应注册读事件、写事件、注销读事件、写事件、清空事件。hiredis会在需要时调用这些回调函数。例如，当我们调用redisAsyncCommand时，hiredis会注册调用addWrite回调函数，然后在用户代码中，我们再epoll中注册写事件，当写事件触发时，调用我们设置的write_fn。

   read_fn和write_fn被保存于event_t中，也是需要用户定义和设置：

   static void redisReadHandler(int fd, int events, void *privdata) {((void)fd);((void)events);printf("redisReadHandler %d\n", fd);event_t *e = (event_t*)privdata;redis_event_t *re = (redis_event_t *)(char *)e;redisAsyncHandleRead(re->ctx);
   }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);
   }
   

   在上面的回调函数中，调用了redisAsyncHandleRead和redisAsyncHandleWrite这两个hiredis提供的API来将数据从内核读写到用户态的buffer中，这个buffer也是hiredis提供的。redisAsyncHandleWrite函数读取到一个完整的响应后会调用用户在调用redisAsyncCommand时设置的回调函数。

   我们可以选择在注册读写事件的时候设置读写事件发生时的回调函数，当然应该也可以选择在创建redis_event_t的时候就设置。

   static void redisAddRead(void *privdata) {redis_event_t *re = (redis_event_t *)privdata;re->e.read_fn = redisReadHandler;redisEventUpdate(privdata, EPOLLIN, 0);
   }
   static void redisAddWrite(void *privdata) {redis_event_t *re = (redis_event_t *)privdata;re->e.write_fn = redisWriteHandler;redisEventUpdate(privdata, EPOLLOUT, 0);
   }
   

完整代码：

#include <hiredis/hiredis.h>
#include <hiredis/async.h>
#include <hiredis/sds.h>#include "reactor.h"
#include "adapter_async.h"static reactor_t *R;char *rtype[] = {"^o^","STRING","ARRAY","INTEGER","NIL","STATUS","ERROR","DOUBLE","BOOL","MAP","SET","ATTR","PUSH","BIGNUM","VERB",
};void dumpReply(struct redisAsyncContext *c, void *r, void *privdata) {redisReply *reply = (redisReply*)r;switch (reply->type) {case REDIS_REPLY_STATUS:case REDIS_REPLY_STRING:printf("[req = %s]reply:(%s)%s\n", (char*)privdata, rtype[reply->type], reply->str);break;case REDIS_REPLY_NIL:printf("[req = %s]reply:(%s)nil\n", (char*)privdata, rtype[reply->type]);break;case REDIS_REPLY_INTEGER:printf("[req = %s]reply:(%s)%lld\n", (char*)privdata, rtype[reply->type], reply->integer);break;case REDIS_REPLY_ARRAY:printf("[req = %s]reply(%s):number of elements=%lu\n", (char*)privdata, rtype[reply->type], reply->elements);for (size_t i = 0; i < reply->elements; i++) {printf("\t %lu : %s\n", i, reply->element[i]->str);}break;case REDIS_REPLY_ERROR:printf("[req = %s]reply(%s):err=%s\n", (char*)privdata, rtype[reply->type], reply->str);break;default:printf("[req = %s]reply(%s)\n", (char*)privdata, rtype[reply->type]);break;}
}void
connectCallback(const redisAsyncContext *c, int status) {if (status != REDIS_OK) {printf("Error: %s\n", c->errstr);stop_eventloop(R);return;}printf("Connected...\n");redisAsyncCommand((redisAsyncContext *)c, dumpReply, "hmset role:10001", "hmset role:10001 name mark age 31 sex male");int a = 10;redisAsyncCommand((redisAsyncContext *)c, dumpReply, "hgetall role:10001", "hgetall role:10001");// ....
}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) {R = create_reactor();redisAsyncContext *c = redisAsyncConnect("127.0.0.1", 6379);if (c->err) {/* Let *c leak for now... */printf("Error: %s\n", c->errstr);return 1;}redisAttach(R, c);redisAsyncSetConnectCallback(c, connectCallback);redisAsyncSetDisconnectCallback(c, disconnectCallback);eventloop(R);release_reactor(R);return 0;
}

学习参考

学习更多相关知识请参考零声 github。