017.Redis Cluster请求路由原理-待完善

CoderJed

发布于 2020-03-27 10:26:41

2.4K0

发布于 2020-03-27 10:26:41

文章被收录于专栏：Jed的技术阶梯Jed的技术阶梯

Redis集群对客户端通信协议做了比较大的修改，为了追求性能最大化，并没有采用代理的方式而是采用客户端直连节点的方式。因此从单机切换到集群环境的应用，需要修改客户端代码。

1. 请求重定向

在集群模式下，Redis接收任何键相关命令时首先计算键对应的槽，再根据槽找出所对应的节点，如果节点是自身，则处理键命令；否则回复MOVED重定向错误，通知客户端请求正确的节点。这个过程称为MOVED重定向。

# 如果key经过计算后，其分配的slot就在当前节点，那么可以请求成功，否则，回复重定向消息
[root@node01 redis]# redis-cli -h 10.0.0.100 -p 6379
10.0.0.100:6379> set name tom
OK
10.0.0.100:6379> set age 20
(error) MOVED 741 10.0.0.101:6379

重定向信息包含了键所对应的槽以及负责该槽的节点地址，根据这些信息客户端就可以向正确的节点发起请求。在10.0.0.101:6379节点上成功执行之前的命令：

[root@node02 redis]# redis-cli -h 10.0.0.101 -p 6379
10.0.0.101:6379> set age 20
OK

使用redis-cli命令时，可以加入-c参数支持自动重定向，简化手动发起重定向的操作：

[root@node01 redis]# redis-cli -c -h 10.0.0.100 -p 6379
10.0.0.100:6379> set age 30
-> Redirected to slot [741] located at 10.0.0.101:6379
OK

redis-cli自动帮我们连接到正确的节点执行命令，这个过程是在redis-cli内部维护，实质上是client端接到MOVED信息指定的节点之后再次发起请求，并不是在当前Redis节点中完成请求转发，节点对于不属于它的键命令只回复重定向响应，并不负责转发。

键命令执行步骤主要分两步：

计算槽

Redis首先需要计算键所对应的槽，根据键的有效部分使用CRC16函数计算出散列值，再取对16383的余数，得到槽的编号，这样每个键都可以映射到0~16383槽范围内

10.0.0.101:6379> cluster keyslot age
(integer) 741

Redis集群相对单机在功能上存在一些限制，限制如下：

key批量操作支持有限，如mset、mget，目前只支持具有相同slot值的key执行批量操作。对于映射为不同slot值的key由于执行mget、mget等操作可能存在于多个节点上因此不被支持
key事务操作支持有限，同理只支持多key在同一节点上的事务操作，当多个key分布在不同的节点上时无法使用事务功能
key作为数据分区的最小粒度，因此不能将一个大的键值对象如hash、list等映射到不同的节点
不支持多数据库空间，单机下的Redis可以支持16个数据库，集群模式下只能使用一个数据库空间，即db0
复制结构只支持一层，从节点只能复制主节点，不支持嵌套树状复制结构

10.0.0.102:6379> mget name age
(error) CROSSSLOT Keys in request don't hash to the same slot

但通常会有这样的需求，例如把一个用户的信息存入到一个slot中，这是可以这样设置：

10.0.0.102:6379> set user:{user1}:name tony
-> Redirected to slot [8106] located at 10.0.0.100:6379
OK
10.0.0.100:6379> set user:{user1}:age 20
OK

10.0.0.100:6379> cluster keyslot user:{user1}:name
(integer) 8106
10.0.0.100:6379> cluster keyslot user:{user1}:age
(integer) 8106

10.0.0.100:6379> mget user:{user1}:name user:{user1}:age
1) "tony"
2) "20"

这样，这两个key在计算hash值的时候，不会根据整个key来计算，而是只是拿{}中的内容的来计算，这样它们的hash值一定是相同的，就可以分配到同一个slot中，{}中的内容称为hash_tag

查找槽所对应的节点

Redis计算得到键对应的槽后，需要查找槽所对应的节点。集群内通过消息交换每个节点都会知道所有节点的槽信息。

根据MOVED重定向机制，客户端可以随机连接集群内任一Redis获取键所在节点，这种客户端又叫Dummy(傀儡)客户端，它优点是代码实现简单，对客户端协议影响较小，只需要根据重定向信息再次发送请求即可。但是它的弊端很明显，每次执行键命令前都要到Redis上进行重定向才能找到要执行命令的节点，额外增加了IO开销，这不是Redis集群高效的使用方式。正因为如此通常集群客户端都采用另一种实现：Smart客户端

2. Smart客户端

大多数开发语言的Redis客户端都采用Smart客户端支持集群协议，Smart客户端通过在内部维护slot→node的映射关系，本地就可实现键到节点的查找，从而保证IO效率的最大化，而MOVED重定向负责协助Smart客户端更新slot→node映射。下面以Jedis为例，说明Smart客户端操作集群的流程

2.1 JedisCluster操作集群的流程

首先在JedisCluster初始化时会选择一个运行节点，初始化槽和节点映射关系，使用cluster slots命令完成

10.0.0.102:6379> cluster slots
1) 1) (integer) 0
   2) (integer) 5460
   3) 1) "10.0.0.101"
      2) (integer) 6379
      3) "95c88177dda69da2e423e9704e1866b3eafc863d"
   4) 1) "10.0.0.100"
      2) (integer) 6380
      3) "20bc884aa53a37b6923c1c18373d6d7e07961f56"
2) 1) (integer) 10923
   2) (integer) 16383
   3) 1) "10.0.0.102"
      2) (integer) 6379
      3) "7b6284343423821db6343887528bf3e0e0a706e6"
   4) 1) "10.0.0.101"
      2) (integer) 6380
      3) "e417b330230eec2c54c3209a4d005b839d37e582"
3) 1) (integer) 5461
   2) (integer) 10922
   3) 1) "10.0.0.100"
      2) (integer) 6379
      3) "9c02aef2d45e44678202721ac923c615dd8300ea"
   4) 1) "10.0.0.102"
      2) (integer) 6380
      3) "0955dc1eeeec59c1e9b72eca5bcbcd04af108820"

JedisCluster解析cluster slots结果缓存在本地，并为每个节点创建唯一的JedisPool连接池。映射关系在JedisClusterInfoCache类中

 public class JedisClusterInfoCache {
    private final Map<String, JedisPool> nodes;
    private final Map<Integer, JedisPool> slots;
    .....
}

JedisCluster执行命令的过程

public abstract class JedisClusterCommand<T> {

  // 集群节点连接处理器
  private JedisClusterConnectionHandler connectionHandler;
  // 重试次数，默认5次
  private int redirections;
  private ThreadLocal<Jedis> askConnection = new ThreadLocal<Jedis>();

  // 模板回调方法
  public abstract T execute(Jedis connection);

  // 利用重试机制运行命令
  private T runWithRetries(byte[] key, int redirections, boolean tryRandomNode, boolean asking) {
    if (redirections <= 0) {
      throw new JedisClusterMaxRedirectionsException("Too many Cluster redirections?");
    }

    Jedis connection = null;
    try {

      if (asking) {
        // TODO: Pipeline asking with the original command to make it
        // faster....
        connection = askConnection.get();
        connection.asking();

        // if asking success, reset asking flag
        asking = false;
      } else {
        if (tryRandomNode) {
          // 随机获取活跃节点连接
          connection = connectionHandler.getConnection();
        } else {
          // 使用 slot 缓存获取目标节点连接
          connection = connectionHandler.getConnectionFromSlot(JedisClusterCRC16.getSlot(key));
        }
      }

      return execute(connection);
    } catch (JedisConnectionException jce) {
      if (tryRandomNode) {
        // maybe all connection is down
        throw jce;
      }

      // release current connection before recursion
      releaseConnection(connection);
      connection = null;

      // retry with random connection
      // 出现连接错误使用随机连接重试
      return runWithRetries(key, redirections - 1, true, asking);
    } catch (JedisRedirectionException jre) {
      // if MOVED redirection occurred,
      if (jre instanceof JedisMovedDataException) {
        // it rebuilds cluster's slot cache
        // recommended by Redis cluster specification
        // 如果出现 MOVED 重定向错误 , 在连接上执行 cluster slots 命令重新初始化 slot 缓存
        this.connectionHandler.renewSlotCache(connection);
      }

      // release current connection before recursion or renewing
      releaseConnection(connection);
      connection = null;

      if (jre instanceof JedisAskDataException) {
        asking = true;
        askConnection.set(this.connectionHandler.getConnectionFromNode(jre.getTargetNode()));
      } else if (jre instanceof JedisMovedDataException) {
      } else {
        throw new JedisClusterException(jre);
      }

      // slot 初始化后重试执行命令
      // 每次命令重试对redirections参数减1
      return runWithRetries(key, redirections - 1, false, asking);
    } finally {
      releaseConnection(connection);
    }
  }

  private void releaseConnection(Jedis connection) {
    if (connection != null) {
      connection.close();
    }
  }

}

整个流程为：
- 计算slot并根据slots缓存获取目标节点连接，执行命令
- 如果出现连接错误，使用随机连接重新执行命令，每次命令重试对redirections参数减1
- 捕获到MOVED重定向错误，使用cluster slots命令更新slots缓存
- 重复执行前3步，直到命令执行成功，或者当redirections<=0时抛出Jedis ClusterMaxRedirectionsException异常

2.2 Jedis-2.8.2之前的代码问题

从命令执行流程中发现，客户端需要结合异常和重试机制时刻保证跟Redis集群的slots同步，因此Smart客户端相比单机客户端有了很大的变化和实现难度。了解命令执行流程后，下面我们对Smart客户端成本和可能存在的问题进行分析

客户端内部维护slots缓存表，并且针对每个节点维护连接池，当集群规模非常大时，客户端会维护非常多的连接并消耗更多的内存。
使用Jedis操作集群时最常见的错误是throw new JedisClusterMaxRedirectionsException("Too many Cluster redirections?")，这经常会引起开发人员的疑惑，它隐藏了内部错误细节，原因是节点宕机或请求超时都会抛出JedisConnectionException，导致触发了随机重试，当重试次数耗尽抛出这个错误
当出现JedisConnectionException时，Jedis认为可能是集群节点故障需要随机重试来更新slots缓存，因此了解哪些异常将抛出JedisConnectionException变得非常重要，有如下几种情况会抛出JedisConnectionException
- Jedis连接节点发生socket错误时抛出
- 所有命令/Lua脚本读写超时抛出
- JedisPool连接池获取可用Jedis对象超时抛出
前两点都可能是节点故障需要通过JedisConnectionException来更新slots缓存，但是第三点没有必要，因此Jedis2.8.1版本之后对于连接池的超时抛出Jedis Exception，从而避免触发随机重试机制
Redis集群支持自动故障转移，但是从故障发现到完成转移需要一定的时间，节点宕机期间所有指向这个节点的命令都会触发随机重试，每次收到MOVED重定向后会调用JedisClusterInfoCache类的discoverClusterSlots()方法

private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
private final Lock r = rwl.readLock();
private final Lock w = rwl.writeLock();

public void discoverClusterSlots(Jedis jedis) {
    // 获取写锁
    w.lock();

    try {
      this.slots.clear();

      // 执行cluster slots命令获取节点和slot映射列表
      List<Object> slots = jedis.clusterSlots();

      for (Object slotInfoObj : slots) {
        // ....初始化slots缓存的Map
      }
    } finally {
      w.unlock();
    }
  }

public JedisPool getSlotPool(int slot) {
    // 获取读锁
    r.lock();
    try {
      // 返回slot对应的jedisPool
      return slots.get(slot);
    } finally {
      r.unlock();
    }
  }

从代码中看到，获得写锁后再执行cluster slots命令初始化缓存，由于集群所有的键命令都会执行getSlotPool()方法计算槽对应节点，它内部要求读锁。Reentrant ReadWriteLock是读锁共享且读写锁互斥，从而导致所有的请求都会造成阻塞。对于并发量高的场景将极大地影响集群吞吐。这个现象称为cluster slots风暴，有如下现象：

重试机制导致IO通信放大问题。比如默认重试5次的情况，当抛出JedisClusterMaxRedirectionsException异常时，内部最少需要9次IO通信：5次发送命令+2次ping命令保证随机节点正常+2次cluster slots命令初始化slots缓存，导致异常判定时间变长
个别节点操作异常导致频繁的更新slots缓存，多次调用cluster slots命令，高并发时将过度消耗Redis节点资源，如果集群slot<->node映射庞大则cluster slots返回信息越多，问题越严重。
频繁触发更新本地slots缓存操作，内部使用了写锁，阻塞对集群所有的键命令调用

2.3 Jedis-2.8.2+的改进

当接收到JedisConnectionException时不再轻易初始化slots缓存，大幅降低内部IO次数

private T runWithRetries(final int slot, int attempts, boolean tryRandomNode, JedisRedirectionException redirect) {
    if (attempts <= 0) {
      throw new JedisClusterMaxAttemptsException("No more cluster attempts left.");
    }

    Jedis connection = null;
    try {

      if (redirect != null) {
        // 获取连接
        connection = this.connectionHandler.getConnectionFromNode(redirect.getTargetNode());
        if (redirect instanceof JedisAskDataException) {
          // TODO: Pipeline asking with the original command to make it faster....
          connection.asking();
        }
      } else {
        if (tryRandomNode) {
          connection = connectionHandler.getConnection();
        } else {
          connection = connectionHandler.getConnectionFromSlot(slot);
        }
      }

      return execute(connection);

    } catch (JedisNoReachableClusterNodeException jnrcne) {
      throw jnrcne;
    } catch (JedisConnectionException jce) {
      // release current connection before recursion
      releaseConnection(connection);
      connection = null;

      // 当重试到 1 次时，更新本地 slots 缓存
      if (attempts <= 1) {
        //We need this because if node is not reachable anymore - we need to finally initiate slots
        //renewing, or we can stuck with cluster state without one node in opposite case.
        //But now if maxAttempts = [1 or 2] we will do it too often.
        //TODO make tracking of successful/unsuccessful operations for node - do renewing only
        //if there were no successful responses from this node last few seconds
        this.connectionHandler.renewSlotCache();
      }
      
      // 递归执行重试
      return runWithRetries(slot, attempts - 1, tryRandomNode, redirect);
    } catch (JedisRedirectionException jre) {
      // if MOVED redirection occurred,
      // 如果是 MOVED 异常，更新 slots 缓存
      if (jre instanceof JedisMovedDataException) {
        // it rebuilds cluster's slot cache recommended by Redis cluster specification
        this.connectionHandler.renewSlotCache(connection);
      }

      // release current connection before recursion
      releaseConnection(connection);
      connection = null;

      // 递归，执行重试
      return runWithRetries(slot, attempts - 1, false, jre);
    } finally {
      releaseConnection(connection);
    }
  }

当更新slots缓存时，不再使用ping命令检测节点活跃度，并且使用redis covering变量保证同一时刻只有一个线程更新slots缓存，其他线程忽略，优化了写锁阻塞和cluster slots调用次数

public void renewClusterSlots(Jedis jedis) {
    //If rediscovering is already in process - no need to start one more same rediscovering, just return
    // 使用rediscovering变量保证当有一个线程正在初始化slots时，其他线程直接忽略
    if (!rediscovering) {
      try {
        w.lock();
        if (!rediscovering) {
          rediscovering = true;

          try {
            if (jedis != null) {
              try {
                // 更新本地缓存
                discoverClusterSlots(jedis);
                return;
              } catch (JedisException e) {
                //try nodes from all pools
                // 忽略异常，使用随机查找更新slots
              }
            }

            // 使用随机查找更新slots
            for (JedisPool jp : getShuffledNodesPool()) {
              Jedis j = null;
              try {
                // 不再使用ping命令检测节点
                j = jp.getResource();
                discoverClusterSlots(j);
                return;
              } catch (JedisConnectionException e) {
                // try next nodes
              } finally {
                if (j != null) {
                  j.close();
                }
              }
            }
          } finally {
            // 释放锁和rediscovering变量
            rediscovering = false;      
          }
        }
      } finally {
        w.unlock();
      }
    }
  }

综上所述，Jedis2.8.2之后的版本，当出现JedisConnectionException时，命令发送次数变为5次：4次重试命令+1次cluster slots命令，同时避免了cluster slots不必要的并发调用。

建议升级到Jedis2.8.2以上版本防止cluster slots风暴和写锁阻塞问题，但还可以进一步优化，如下所示：

执行cluster slots的过程不需要加入任何读写锁，因为cluster slots命令执行不需要做并发控制，只有修改本地slots时才需要控制并发，这样降低了写锁持有时间
当获取新的slots映射后使用读锁跟老slots比对，只有新老slots不一致时再加入写锁进行更新。防止集群slots映射没有变化时进行不必要的加写锁行为

3. ASK重定向

3.1 客户端ASK重定向流程

当slot对应的数据从源节点到目标节点迁移过程中，客户端需要做到智能识别，保证键命令可正常执行。例如当一个slot数据从源节点迁移到目标节点时，期间可能出现一部分数据在源节点，而另一部分在目标节点，当出现上述情况时，客户端键命令执行流程将发生变化。

客户端根据本地slots缓存发送命令到源节点，如果存在键对象则直接执行并返回结果给客户端
如果键对象不存在，则可能存在于目标节点，这时源节点会回复ASK重定向异常。格式如下： (error) ASK {slot} {targetIP}:{targetPort}
客户端收到重定向指令之后，先去目标节点执行一个不带参数的asking指令，然后在目标节点执行原来的操作请求指令
客户端先执行一个不带参数的asking指令的原因是，在迁移完成之前，按道理来说，这个slot还是不属于目标节点的，于是目标节点会给客户端返回-MOVED指令，让客户端去原节点执行操作，这样就形成了"互相推托"的重定向循环。
asking指令就是告诉目标节点，"我的指令你必须处理，请求的slot就当成是你的吧"
如果数据在目标节点存在则执行命令，不存在则返回不存在信息
迁移会影响服务效率，在正常情况下，一次请求就可以完成操作，而在迁移过程中，客户端需要请求3次(发送给原节点、发送给目标节点asking指令，发送给目标节点真正的处理请求)

ASK与MOVED虽然都是对客户端的重定向控制，但是有着本质区别，ASK重定向说明集群正在进行slot数据迁移，客户端无法知道什么时候迁移完成，因此只能是临时性的重定向，客户端不会更新slots缓存，但是MOVED重定向说明键对应的槽已经明确指定到新的节点，因此需要更新slots缓存

slot迁移感知

如果某个slot已经迁移完了，那么客户端如何才能感知到slot的变化呢？客户端包存了slot和节点的映射关系表，它需要得到及时更新，才可以正常的将请求发送到正确的节点上

Redis Cluster是去中心化的，客户端只要访问其中的一个节点就可以，其他的节点通过访问的这个节点来获取，当然也可以同时提供多个节点，这样安全性会更好，如果只提供一个地址，那么当这个节点挂了，客户就必须更换访问地址才可以继续访问Cluster
当客户端发送一个请求到节点1，节点发现这个slot并不是由自己管理，于是给客户端返回-MOVED 3999 127.0.0.1:6381响应，3999是客户端要访问的key的slot的编号，后面就是该slot所在的目标节点，-MOVED中的减号则是代表客户端的这个请求是错误的，这样就告诉了客户端："slot不再我这儿，你去127.0.0.1:6381找3999号slot"
于是客户端根据-MOVED的提示，将请求打到真正的目标节点，此时如果目标节点正在迁移过程中，请求又会被转到原先slot所在的旧节点，如果旧节点有数据，那么就直接返回结果了，如果旧节点中没有此条数据，那么旧节点又会让客户端去目标节点拿数据，这时候旧节点返回给客户端一个asking error响应并携带着目标节点的地址，客户端又一次跑到目标节点那里去尝试获取数据，在这样的过程中，客户端不会刷新自己的槽位映射表，因为这只是临时的纠正槽位信息。
这样的过程不会无限制的循环下去，用户可以通过参数指定重试的次数，当超过这个次数还没有获取到数据，就会抛出异常

客户端在redis集群中请求正在迁移的slot的流程：