codis的Rebalance算法

Codis 由四部分组成:

• Codis Proxy (codis-proxy)
• Codis Dashboard (codis-config)
• Codis Redis (codis-server)
• ZooKeeper/Etcd

 

codis-proxy 是客户端链接的 Redis 代理服务, codis-proxy 自己实现了 Redis 协议, 表现得和一个原生的 Redis 没什么区别 (就像 Twemproxy), 对于一个业务来讲, 能够部署多个 codis-proxy, codis-proxy 自己是无状态的.

 

codis-config 是 Codis 的管理工具, 支持包括, 添加/删除 Redis 节点, 添加/删除 Proxy 节点, 发起数据迁移等操做. codis-config 自己还自带了一个 http server, 会启动一个 dashboard, 用户能够直接在浏览器上观察 Codis 集群的运行状态.

codis-server 是 Codis 项目维护的一个 Redis 分支, 基于 2.8.21 开发, 加入了 slot 的支持和原子的数据迁移指令. Codis 上层的 codis-proxy 和 codis-config 只能和这个版本的 Redis 交互才能正常运行.

Codis 依赖 ZooKeeper 来存放数据路由表和 codis-proxy 节点的元信息, codis-config 发起的命令都会经过 ZooKeeper 同步到各个存活的 codis-proxy.

 

上面这张图的意思是，client能够直接访问codis-proxy，也能够经过访问HAProxy。

由于codis的proxy是无状态的，能够比较容易的搭多个proxy来实现高可用性并横向扩容。 对Java用户来讲，可使用通过咱们修改过的Jedis，Jodis ，来实现proxy层的HA。它会经过监控zk上的注册信息来实时得到当前可用的proxy列表，既能够保证高可用性，也能够经过轮流请求全部的proxy实现负载均衡。若是须要异步请求，可使用咱们基于Netty开发的Nedis。

 

codis的Rebalance

codis的存储层能够实现扩容，下面说明一下codis增长group的策略：

codis是用golang编写，使用了c的一些库，Rebalance的源码以下，仔细看过源码后咱们能清楚理解它的策略

func Rebalance() error {

    targetQuota, err := getQuotaMap(safeZkConn)

    if err != nil {

        return errors.Trace(err)

    }

    livingNodes, err := getLivingNodeInfos(safeZkConn)

    if err != nil {

        return errors.Trace(err)

    }

    log.Infof("start rebalance")

    for _, node := range livingNodes {

        for len(node.CurSlots) > targetQuota[node.GroupId] {

            for _, dest := range livingNodes {

                if dest.GroupId != node.GroupId && len(dest.CurSlots) < targetQuota[dest.GroupId] && len(node.CurSlots) > targetQuota[node.GroupId] {

                    slot := node.CurSlots[len(node.CurSlots)-1]

                    // create a migration task

                    info := &MigrateTaskInfo{

                        Delay:      0,

                        SlotId:     slot,

                        NewGroupId: dest.GroupId,

                        Status:     MIGRATE_TASK_PENDING,

                        CreateAt:   strconv.FormatInt(time.Now().Unix(), 10),

                    }

                    globalMigrateManager.PostTask(info)



                    node.CurSlots = node.CurSlots[0 : len(node.CurSlots)-1]

                    dest.CurSlots = append(dest.CurSlots, slot)

                }

            }

        }

    }

    log.Infof("rebalance tasks submit finish")

    return nil

}

首先getQuotaMap(safeZkConn)获取并计算出存活的group的slot配额，数据结构为map<group, slot_num>。若是新增了group，那么新的配额会在这个函数中计算完成。getLivingNodeInfos(safeZkConn)，这个函数获取存活的group列表。其实getQuotaMap也调用了这个方法，从ZK上获取元数据。下面看一下getQuotaMap(safeZkConn)：

func getQuotaMap(zkConn zkhelper.Conn) (map[int]int, error) {

    nodes, err := getLivingNodeInfos(zkConn)

    if err != nil {

        return nil, errors.Trace(err)

    }

    ret := make(map[int]int)

    var totalMem int64

    totalQuota := 0

    for _, node := range nodes {

        totalMem += node.MaxMemory

    }

    for _, node := range nodes {

        quota := int(models.DEFAULT_SLOT_NUM * node.MaxMemory * 1.0 / totalMem)

        ret[node.GroupId] = quota

        totalQuota += quota

    }

    // round up

    if totalQuota < models.DEFAULT_SLOT_NUM {

        for k, _ := range ret {

            ret[k] += models.DEFAULT_SLOT_NUM - totalQuota

            break

        }

    }

    return ret, nil

}

先遍历nodes，计算最大内存总和，再遍历nodes，根据该节点占总内存大小的比例，分配slot数量，存入返回结果中。若是最后totalQuota数量小于1024，把剩余的solt分配给第一个group。下面是getLivingNodeInfos(safeZkConn)：

func getLivingNodeInfos(zkConn zkhelper.Conn) ([]*NodeInfo, error) {

    groups, err := models.ServerGroups(zkConn, globalEnv.ProductName())

    if err != nil {

        return nil, errors.Trace(err)

    }

    slots, err := models.Slots(zkConn, globalEnv.ProductName())

    slotMap := make(map[int][]int)

    for _, slot := range slots {

        if slot.State.Status == models.SLOT_STATUS_ONLINE {

            slotMap[slot.GroupId] = append(slotMap[slot.GroupId], slot.Id)

        }

    }

    var ret []*NodeInfo

    for _, g := range groups {

        master, err := g.Master(zkConn)

        if err != nil {

            return nil, errors.Trace(err)

        }

        if master == nil {

            return nil, errors.Errorf("group %d has no master", g.Id)

        }

        out, err := utils.GetRedisConfig(master.Addr, globalEnv.Password(), "maxmemory")

        if err != nil {

            return nil, errors.Trace(err)

        }

        maxMem, err := strconv.ParseInt(out, 10, 64)

        if err != nil {

            return nil, errors.Trace(err)

        }

        if maxMem <= 0 {

            return nil, errors.Errorf("redis %s should set maxmemory", master.Addr)

        }

        node := &NodeInfo{

            GroupId:   g.Id,

            CurSlots:  slotMap[g.Id],

            MaxMemory: maxMem,

        }

        ret = append(ret, node)

    }

    cnt := 0

    for _, info := range ret {

        cnt += len(info.CurSlots)

    }

    if cnt != models.DEFAULT_SLOT_NUM {

        return nil, errors.Errorf("not all slots are online")

    }

    return ret, nil

}

这段代码中先获取group和slot信息列表，遍历slots，要求slot是在线状态，若是有任何一个不在线，在最后判断slot数量的时候都会报错。遍历groups，要求任何一个group都有master、配置了maxMem。

继续分析Rebalance。三层遍历，遍历存活节点，若是当前配额大于rebalance以后的配额的话，第三个for循环中标红的判断的结果必定是新增的group。下面的逻辑就是将node中多出来的slot(当前旧的group)迁移到dest(当前新的group)。起一个task，完成迁移。

综上所述，codis的Rebalance算法简单、清晰，符合常规思路。另外go语言的可读性也很强，以前没有涉及过go，但适应一下能够看懂。

若是group中master宕机，须要注意，codis将其中一个slave升级为master时，该组内其余slave实例是不会自动改变状态的，这些slave仍将试图从旧的master上同步数据，于是会致使组内新的master和其余slave之间的数据不一致。由于redis的slave of命令切换master时会丢弃slave上的所有数据，重新master完整同步，会消耗master资源。所以建议在知情的状况下手动操做。使用 codis-config server add <group_id> <redis_addr> slave 命令刷新这些节点的状态便可。codis-ha不会自动刷新其余slave的状态。