谈谈对 go-redis 客户端的一些优化

这篇文章主要是记录一下我们在使用 go-redis 过程中遇到的一些问题及我们的解决方案，主要内容如下：

• 背景
• 部署模式
• 业务需求
• 主要优化点
• 主从复制机制

背景

当前我们业务上主要使用 Go 语言，Redis 使用的则是开源的 go-redis 客户端。但是当前 go-redis 有一个很大的不足就是：在 sentinel 部署模式下，它默认总是获取主库连接，因此在高并发尤其是读多写少的场景下并不适用。具体来说，在我们的场景中，高峰期 QPS 超过 10w，其中 90% 以上都是读请求。且不说 redis 单机能不能抗住这么大的流量，即使暂时能抗住，考虑到业务发展或突发情况，这始终是一个极大的不确定性。因此改造 go-redis 客户端，使其支持读写分离势在必行。

在这之前，我们先简单介绍一下 redis 的几种部署模式。

部署模式

standalone

standalone 就是我们常说的单机模式，所有数据和请求都指向一个 redis 实例，无法保证高可用，一般不建议用于生产环境，因为一旦 redis 出现宕机的情况，就可能导致数据丢失或者引起雪崩。

cluster

为了实现生产环境的高可用，集群部署是我们最容易想到的方案，这就是我们要说的 cluster 模式。与我们常见的集群模式不同，redis 的 cluster 模式类似于一种去中心化的方式。具体来说，cluster 模式下集群有 16384 个哈希槽分布在多个主节点上 ，每个 key 通过 CRC16 校验后对 16384 取模来决定放在哪个槽，因此每个主节点仅保存整个集群的一部分数据。

基于这种去中心化的策略，再在每个主节点挂一个或多个从节点，cluster 模式就能够保证一定程度的高可用。

MasterA、MasterB、MasterC 分别保存了集群的一部分数据，三者数据并集是集群全量数据。

image

但是 cluster 模式也有两个明显的不足：

• cluster 不能保证数据的强一致性。

假设发生网络分区，主节点A、B和从节点A1、B1、C1在一个分区，而主节点 C 和客户端 client 在另外一个分区。此时 client 仍能够向 C 写入数据。如果时间较长，在另外一个分区，从节点C1就会被选举为主节点，此时client 之前写入C的数据就会丢失。

• cluster 不支持处理多个 key。

为了保证多个命令的原子性，我们通常会使用 lua 脚本来实现。但 redis 要求单个 lua 脚本的所有 key 在同一个槽位。虽然 redis 也提供了相应的解决方案，但是对于老业务来说，无法做到平滑迁移。同样，pipeline 和 事务也有类似的问题。

更多详细信息，详见官方文档^[1]

sentinel

sentinel 模式也是 redis 官方推荐的高可用方案。sentinel 模式最典型的特征是提供了一整套完整的监控功能、可以做到 7*24 小时不间断监控、自动故障转移以及作为配置提供者。

MasterA、MasterB 是两个独立的集群，它们保存了各自集群的全量数据。

image

一个 sentinel 集群可以监控多个 master-slave 集群，每个 master-slave 集群有一个主节点和多个从节点，一旦 sentinel 监控到某个 master 宕机，会自动从它的多个 slave 节点中选出最合适的一个作为新的 master 节点。

更多详细信息，详见官方文档^[2]

业务需求

前面已经简单介绍过我们的业务背景，为了支持 12w 的QPS，必须支持读写分离，这个 go-redis 客户端当前已经支持。但因为我们业务上大量使用到了 pipeline 和 lua 脚本，因此如果切回 cluster 模式，我们没办法做到平滑迁移，必须删除大量旧 key。所以改造 go-redis sentinel 的客户端实现，使其支持读写分离似乎是更合适的选择。

我们所有的改造基于 go-redis v7.3.0 版本。

主要优化点

支持读写分离

首先，开放出一个 readOnlySentinel 参数，每次 dial 的过程中，根据该参数决定获取主节点地址还是从节点地址。

获取从节点地址的主要过程如下：

• 利用 sentinel slaves ${masterName}，获取当前集群的所有从节点地址•根据 flags 状态过滤掉没有 ready 的从节点
• 从活跃的从节点中随机选一个作为结果返回
• 同时订阅主题为 "+sdown" 的消息，一旦某个从节点宕机，我们可以及时从该消息中获取相关信息，并将它从连接池中移除

但测试过程中发现，虽然主库请求下来了，但是所有读请求每次都是路由到同一台从库上。分析发现，这是由于 go-redis 的连接池实现导致的，具体如下：

func (p *ConnPool) popIdle() *Conn {
    if len(p.idleConns) == 0 {
        return nil
    }

    idx := len(p.idleConns) - 1
    cn := p.idleConns[idx]
    p.idleConns = p.idleConns[:idx]
    p.idleConnsLen--
    p.checkMinIdleConns()
    return cn
}

func (p *ConnPool) Put(cn *Conn) {
    if cn.rd.Buffered() > 0 {
        internal.Logger.Printf("Conn has unread data")
        p.Remove(cn, BadConnError{})
        return
    }

    if !cn.pooled {
        p.Remove(cn, nil)
        return
    }

    p.connsMu.Lock()
    p.idleConns = append(p.idleConns, cn)
    p.idleConnsLen++
    p.connsMu.Unlock()
    p.freeTurn()
}

func (c *baseClient) releaseConn(cn *pool.Conn, err error) {
    if c.opt.Limiter != nil {
        c.opt.Limiter.ReportResult(err)
    }

    if isBadConn(err, false) {
        c.connPool.Remove(cn, err)
    } else {
        c.connPool.Put(cn)
    }
}

可以看到，put() 时，连接被追加在一个数组里，每次 get() 则从连接数组里取最后一个连接，然后每次 release() 的时候又追加在数组最后一个位置，类似于栈的实现。因此在 QPS 不是很高的时候，客户端永远取的都是连接池数组里最后一个连接。

连接池这么设计当然也有它的好处，比如说可以避免连接池数组的频繁移动等等，但从负载均衡的角度来说却说不上是一个特别好的设计。

当时为了尽快上线，我们采取了一个比较取巧的办法，就是通过参数 MaxConnAge 设置连接的存活时间为1min，之后就将它销毁创建新连接。当然这会导致频繁的创建连接，但能暂时解决负载均衡不均匀的问题，后面会提到我们对它的第二次优化。

第一版改造完成之后，从监控可以看到，差不多有一半的流量已成功从主节点分流到从节点(之所以主节点还有一半的流量，是因为还有业务没有适配新客户端)。

image

优化负载均衡

经过第一步改造，我们已经基本上支持了读写分离，大大降低了主库的压力。但从监控也可以看出，从库的负载均衡仍然算不上很均匀，同一时刻每个从库的压力差异较大。主要原因我们分析可能有两个：一是单纯的随机算法无法在每个时刻都保证所有从库的负载都非常均衡，只能保证从一个较长的时间范围来看平均负载相对均衡；二是默认的连接池实现可能导致在短时间内大量请求都被分配给了连接池里最后一个连接。

为此，我们还需要对其进行进一步优化。具体优化点有两个：

• get() 连接时，每次取数组里的第一个，然后每次 put() 时还是追加在数组最后一个位置，相当于把连接池从前面说到的的栈实现改成了队列实现。
• 每次 dial 获取从库地址时，不再是从可用从库地址列表里随机选取一个，而是通过一个自增的 slaveIdx 对从库实例个数取模的方式来选取从库地址，保证每个从库实例被选择到的概率是相等的。

类似于这么一个环形，每个客户端依次选取第 1、2、3、4、1 号从库进行连接。

image

来看看二次改造之后的效果，可以看到，从库的负载均衡明显比之前更均匀了。

image

效果似乎不错，但前面也说过，由于配置了 MaxConnAge 参数，会导致频繁的销毁和创建连接。那在我们修改了连接池和 dial() 实现后，是否可以去掉该参数了呢？

去掉 MaxConnAge 参数后，我们发现从库负载均衡结果出现了意料之外的结果。虽然从库之间负载均衡的结果变得比预期的要差的多，但同时还呈现出另外一个特点：不同实例的 QPS 随全站流量平稳波动，流量高的实例流量总是高，流量低的实例一直低。

image

这就很有意思了，为什么会出现这种现象呢？

按道理，现在所有从库实例被选中的概率是相等的，连接池里的连接在不同从库实例上的分布也就应该是均匀的。为了验证这一猜测，我们对连接池里连接的从库地址进行了采样统计，发现实际情况跟我们的猜测并不相同，而是呈现出跟上图的 QPS 类似的很明显的分层特性。

进一步分析每个客户端上连接池里连接的分布情况，发现在同一个客户端上，不同从库实例上的连接分布是很均匀的，最多只相差1个，但是因为线上客户端较多，累加的效果导致了最终比较大的差异。

继续观察，我们还发现了一个特点，就是按照不同实例连接数从多到少排序，发现这个顺序与 sentinel slaves masterName 返回的从库实例顺序一致。看到这里，我们这才恍然大悟：原来针对每一个客户端，我们总是按前面说的1、2、3、4 的顺序来选取从实例，如果连接个数刚好是从库实例的整数倍时，那么此时连接在不同实例间的分布当然就是均匀的。可这么巧合的情况终究是比较少的，绝大多数时候并不是这样，而且连接池内连接的数量一直都在动态变化中，所以当连接总数不是从库实例的整数倍时(比如说如果是6个连接)，那么按照我们的顺序，1、2号从库就会比3、4号从库多一个连接。线上所有客户端加起来，就会导致 sentinel slaves masterName 返回结果中越靠前的从库，其上面的连接就越多。

说起来有点绕，举个例子可能更好理解。假设 redis 集群有 4 个从库，同时我们有 3 个 客户端，每个客户端上分别有 5、6、7 个连接，那么按照我们前面的负载均衡算法，3 个客户端上连接池的连接分布情况分别是：

• 1、2、3、4、1
• 1、2、3、4、1、2
• 1、2、3、4、1、2、3

单独看一个客户端，连接的分布式较为均匀的，但是 3 个客户端累加之后，不难发现 1、2、3、4 号从库上的连接数就变成了 6、5、4、3，如果客户端数量更多，这个差异就更大。而如果 3 个客户端分别从 2、3、4 号从库开始建立连接，那么 3 个客户端上的连接分布情况将变为：

• 2、3、4、1、2
• 3、4、1、2、3、4
• 4、1、2、3、4、1、2

此时1、2、3、4号从库上的连接总数分别为：4、5、4、5，显然，这比之前的6、5、4、3要更为均匀，即使客户端数量再多也不受影响。

发现这一点，这个问题要解决起来也就简单了。即在每次选择从库时，不再固定的每次从 1 号从库开始，而是每次随机从其中一个从库 k 开始，然后按照 k, k+1, k+2, ..., 1, 2, ..., k, ... 这种顺序来路由就可以了。再来看看按照这个思路改造后的结果：

• 不同实例间连接总数分布较为均匀了，最多只相差了 3 个连接

image

• 从库 QPS 也从明显的分层开始收敛，最终不同实例 QPS 差异从 1.5k 左右下降为 0.3k 左右

image

支持平滑扩容

就这样线上稳定运行了几个月，前段时间为了提高我们 redis 集群的高可用性，计划对线上 redis 集群进行扩容。原本以为是很简单的一个操作，结果在每个从库实例加入集群的1min内，线上出现了大量这样的 error 日志：

LOADING Redis is loading the dataset in memory

跟 DBA 确认，是由于从库虽然加入了集群，但是数据同步需要时间，在完成数据同步之前并不能对外提供服务。但此时反应在 flags 字段里的从库状态与正常的从库无异，因此在我们 RandomSlaveAddr() 方法就可能会选中正在同步的从库地址。

为了做进一步确认，我们在本地进行了模拟。具体步骤如下：

1.在本地搭建一个一主两从的 sentinel 集群
2.启动 sentinel 节点
3.启动主节点
4.向主节点写入 100w 个 key，占用内存 1.6G 左右
5.启动一个从节点
6.通过 sentinel slaves mymaster 查看从节点状态，正常
7.此时通过从库访问任意一个 key，都会提示"LOADING Redis is loading the dataset in memory"，大概 20s 后恢复正常

至此，基本确认了问题来源，要解决这个问题大概有这么几种思路：

1.修改 redis 源码，在从节点完成数据同步之前，设置其状态为同步中，并将其设置到 flags 字段，这样我们就可以根据该状态过滤掉在同步中的 slave 节点（事实上，目前对于从库在同步中、同步完成的状态，redis 都有相应的事件发出来，但我们无法根据事件来剔除还未完成同步的节点）
2.每次拿到连接之后先 ping 一下，看是否可以收到 pong 的响应，如果正常，则继续执行，否则丢弃该连接
3.每次通过 dial 方法获取从节点时，先 ping 一下看是否可以收到 pong 的响应，如果正常，则继续执行，否则丢弃该节点

方案 1 因为涉及修改 redis 源码，需要我们对 redis 底层实现十分了解，同时熟悉 C 语言开发，这个我们自问还无法做到十足的把握，风险太大。

方案 2 每次拿到连接之后加一个 ping() 操作，会直接导致线上 QPS 翻倍，因此也不考虑。

方案 3 似乎是可行的，可以在我们最开始支持读写分离获取从节点的代码中，通过 ping 操作剔除坏节点。但是我们唯一的担心在于：由于此时只拿到了节点地址，为了能够执行 ping 操作，我们需要每次都新建一个客户端对象，如果新建连接的操作比较频繁，可能就会创建大量的对象导致内存占用飙升(虽然说这些对象会被 GC 回收，但终究是个隐患，如果线上验证不会引发内存飙升，这个方案其实是可行的)。

那有没有更好的办法，不用执行 ping 命令就能识别出节点状态呢？答案当然是可以的，这还需要从 redis 主从复制机制说起。具体复制机制见文末有详细介绍，这里我们只需要知道主从同步完成之后，从 sentinel 获取到的从库状态中 master-link-status 字段会被设置为 ok 就可以了，通过该字段是否为 ok，我们就可以知道该从库是否完成了与主库的同步。

因此这就解决了我们前面 3 种备选方案的所有问题：

• 不需要修改 redis 源码，只需要修改 go-redis 客户端即可
• 不需要执行 ping 命令，无需担心线上 QPS 大幅增加
• 不需要新建客户端，无需担心内存飙升

事实上，go-redis 还有另外一个钩子方法 OnConnect() 也可以帮助我们实现平滑扩容。在每次有新连接加入连接池之后，都会回调该方法中自定义的操作，如果发生异常则清除掉该连接。只不过其默认是个空实现，因此我们可以在该方法中加入一个 ping 操作：

if option.OnConnect == nil {
    option.OnConnect = func(c *Conn) error {
        return c.conn.Ping().Err()
    }
}

主从复制机制

在支持平滑扩容一节，我们说到可以通过 master-link-status 状态来过滤掉还没完成主从同步的节点。具体原因还需要从 redis 的主从复制机制说起，下面将对该机制进行简单的介绍。

状态机

建立主从关系可以通过 slaveof [masterip] [masterport] 来实现，该命令入口函数是 slaveofCommand()。整个复制过程维护了一个比较复杂的状态机，该状态机的状态如下：

/* Slave replication state. Used in server.repl_state for slaves to remember
 * what to do next. */
#define REPL_STATE_NONE 0 /* No active replication */
#define REPL_STATE_CONNECT 1 /* Must connect to master */
#define REPL_STATE_CONNECTING 2 /* Connecting to master */
/* --- Handshake states, must be ordered --- */
#define REPL_STATE_RECEIVE_PONG 3 /* Wait for PING reply */
#define REPL_STATE_SEND_AUTH 4 /* Send AUTH to master */
#define REPL_STATE_RECEIVE_AUTH 5 /* Wait for AUTH reply */
#define REPL_STATE_SEND_PORT 6 /* Send REPLCONF listening-port */
#define REPL_STATE_RECEIVE_PORT 7 /* Wait for REPLCONF reply */
#define REPL_STATE_SEND_IP 8 /* Send REPLCONF ip-address */
#define REPL_STATE_RECEIVE_IP 9 /* Wait for REPLCONF reply */
#define REPL_STATE_SEND_CAPA 10 /* Send REPLCONF capa */
#define REPL_STATE_RECEIVE_CAPA 11 /* Wait for REPLCONF reply */
#define REPL_STATE_SEND_PSYNC 12 /* Send PSYNC */
#define REPL_STATE_RECEIVE_PSYNC 13 /* Wait for PSYNC reply */
/* --- End of handshake states --- */
#define REPL_STATE_TRANSFER 14 /* Receiving .rdb from master */
#define REPL_STATE_CONNECTED 15 /* Connected to master */

很显然，这么复杂的操作必须异步化。状态机状态设置为 REPL_STATE_CONNECT 后直接返回，后续操作在serverCron() 里实现。redis 里一些需要异步化的操作都会放在这个函数里，其中从库的同步操作每秒执行一次。

/* Replication cron function -- used to reconnect to master,
     * detect transfer failures, start background RDB transfers and so forth. */
run_with_period(1000) replicationCron();

在 replicationCron() 方法里，如果发现状态机状态是 REPL_STATE_CONNECT，就会向主库发起连接，更新状态机状态为 REPL_STATE_CONNECTING。

 /* Check if we should connect to a MASTER */
if (server.repl_state == REPL_STATE_CONNECT) {
    serverLog(LL_NOTICE,"Connecting to MASTER %s:%d",
        server.masterhost, server.masterport);
    if (connectWithMaster() == C_OK) {
        serverLog(LL_NOTICE,"MASTER <-> SLAVE sync started");
    }
}

主从握手

之后是从库和主库之间的一系列握手操作，具体包括：

• 发送 ping 命令给主库，等待 pong 响应（REPL_STATE_RECEIVE_PONG）
• 权限认证（REPL_STATE_SEND_AUTH、REPL_STATE_RECEIVE_AUTH）
• 发送从库端口信息给主库（REPL_STATE_SEND_PORT、REPL_STATE_RECEIVE_IP）•发送从库 IP 信息给主库（REPL_STATE_SEND_IP、REPL_STATE_RECEIVE_IP）
• 通知主库，当前从库已经 ready，具备了处理 RDB 文件的能力（REPL_STATE_SEND_CAPA、REPL_STATE_RECEIVE_CAPA）
• 从库向主库发起同步请求（REPL_STATE_SEND_PSYNC、REPL_STATE_RECEIVE_PSYNC）

至此，从库和主库之间的握手操作完成。然后根据握手结果，决定进行增量同步还是全量同步。

主从同步

在 slaveTryPartialResynchronization 方法里，从库会将自己关联主库的 runid 和自己当前的偏移量 offset 发送给主库：

reply = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"PSYNC",psync_replid,psync_offset,NULL);

然后尝试进行增量同步

if (!strncmp(reply,"+FULLRESYNC",11)) {
    /* FULL RESYNC, parse the reply in order to extract the run id and the replication offset. */
    ...

  return PSYNC_FULLRESYNC;
}

if (!strncmp(reply,"+CONTINUE",9)) {
    /* Partial resync was accepted. */
  ...
  // 开启增量同步
  replicationResurrectCachedMaster(fd)
  return PSYNC_CONTINUE;
}

• 如果增量同步成功，会在 replicationResurrectCachedMaster() 方法里将状态机设置为 REPL_STATE_CONNECTED，主从同步完成
• 如果是全量同步，从库接收主库发送过来的 rdb 文件，更新状态机为 REPL_STATE_TRANSFER

psync_result = slaveTryPartialResynchronization(fd,1);
if (psync_result == PSYNC_WAIT_REPLY) return; /* Try again later... */

if (psync_result == PSYNC_CONTINUE) {
    serverLog(LL_NOTICE, "MASTER <-> SLAVE sync: Master accepted a Partial Resynchronization.");
    return;
}

...
  
/* Prepare a suitable temp file for bulk transfer */
while(maxtries--) {
    snprintf(tmpfile,256,
        "temp-%d.%ld.rdb",(int)server.unixtime,(long int)getpid());
    dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
    if (dfd != -1) break;
    sleep(1);
}

/* Setup the non blocking download of the bulk file. */
if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL)== AE_ERR)
{
    serverLog(LL_WARNING,
        "Can't create readable event for SYNC: %s (fd=%d)", strerror(errno),fd);
    goto error;
}
...
  
server.repl_state = REPL_STATE_TRANSFER;

其中 readSyncBulkPayload() 是从库接收 rdb 文件的处理逻辑，处理成功后会将状态机更新为 REPL_STATE_CONNECTED，这也标志着主从同步的完成。

与此同时，从库会将自己 info 命令中 Replication 的 mater-link-status 字段更新为 up：

info = sdscatprintf(info,
    "master_host:%s\r\n"
    "master_port:%d\r\n"
    "master_link_status:%s\r\n"
    "master_last_io_seconds_ago:%d\r\n"
    "master_sync_in_progress:%d\r\n"
    "slave_repl_offset:%lld\r\n"
    ,server.masterhost,
    server.masterport,
    (server.repl_state == REPL_STATE_CONNECTED) ?
        "up" : "down",
    server.master ?
    ((int)(server.unixtime-server.master->lastinteraction)) : -1,
    server.repl_state == REPL_STATE_TRANSFER,
    slave_repl_offset
);

sentinel 则在返回的 slave 信息中将 master-link-status 字段更新为 "ok"，这也是我们前面提到支持平滑扩容时可以根据该字段状态来识别坏节点的原因。

 /* master_link_status:<status> */
if (sdslen(l) >= 19 && !memcmp(l,"master_link_status:",19)) {
    ri->slave_master_link_status =
        (strcasecmp(l+19,"up") == 0) ?
        SENTINEL_MASTER_LINK_STATUS_UP :
        SENTINEL_MASTER_LINK_STATUS_DOWN;
}

addReplyBulkCString(c,
   (ri->slave_master_link_status == SENTINEL_MASTER_LINK_STATUS_UP) ? "ok" : "err");

那么问题来了：主库是如何判断一个从库到底能不能进行增量同步的呢？

这是由于主库维护了一个复制积压缓冲区 repl_backlog，这是一个 1M 大小的循环队列，所有对主库写入的内容都会同时写入该队列。从库发送同步请求后，主库会优先进行增量同步的尝试，如果从库申请同步的 offset 在该队列范围内，说明可以进行增量同步，否则表示有数据丢失，必须进行全量同步。

int masterTryPartialResynchronization(client *c) {
  ...
  /* We still have the data our slave is asking for? */
  if (!server.repl_backlog ||
      psync_offset < server.repl_backlog_off ||
      psync_offset > (server.repl_backlog_off + server.repl_backlog_histlen)) {
    
        ...
  }
}

这里的 backlog 其实跟 MySQL 的 redolog 很类似，都具备容量有限、循环写入的特点。区别在于在 MySQL 中， redolog 有多个文件，一旦所有 redolog 文件写满，MySQL 将不得不停下所有更新操作来刷脏页，而这里的 backlog 则是直接覆盖前面的内容。

命令传播

主从同步完成之后，后续由于主库数据的更新将会通过命令传播的方式同步到从库，当然这也是一个异步操作。

/* Propagate the specified command (in the context of the specified database id)
 * to AOF and Slaves.
 */
void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
               int flags)
{
    if (server.aof_state != AOF_OFF && flags & PROPAGATE_AOF)
        feedAppendOnlyFile(cmd,dbid,argv,argc);
    if (flags & PROPAGATE_REPL)
        replicationFeedSlaves(server.slaves,dbid,argv,argc);
}

主从心跳

replicationCron 这个定时任务里面还维持了主从之间的心跳机制 ：

• 主库定期向所有从库发送 ping 命令，具体周期通过参数 repl_ping_slave_period(5.0 之后的版本更名为 repl-ping-replica-period) 控制，默认 10s
• 从库通过 replicationSendAck 每隔 1s 向主库发送 ack，上报自己当前的复制偏移量

/* Send ACK to master from time to time.

全文完，感谢阅读。

References

[1] 官方文档: https://redis.io/topics/cluster-tutorial
[2] 官方文档: https://redis.io/topics/sentinel