项目实践，Redis集群技术学习(十)

用户1289394

发布于 2022-02-15 16:00:14

4900

发布于 2022-02-15 16:00:14

文章被收录于专栏：Java学习网

2.忘记节点

由于集群内的节点不停地通过 Gossip 消息彼此交换节点状态，因此需要通过一种健壮的机制让集群内所有节点忘记下线的节点。也就是说让其他节点不再与要下线节点进行 Gossip 消息交换。Redis 提供了 cluster forget{downNodeId}命

令实现该功能。

当节点接收到 cluster forget{down NodeId}命令后，会把 nodeId 指定的节点加入到禁用列表中，在禁用列表内的节点不再发送 Gossip 消息。禁用列表有效期是 60 秒，超过 60 秒节点会再次参与消息交换。也就是说当第一次 forget 命令发出后，我们有 60 秒的时间让集群内的所有节点忘记下线节点。

线上操作不建议直接使用 cluster forget 命令下线节点，需要跟大量节点命令交互，实际操作起来过于繁琐并且容易遗漏 forget 节点。建议使用 redis-trib.rbdel-node{host：port}{downNodeId}命令，

当下线主节点具有从节点时需要把该从节点指向到其他主节点，因此对于主从

节点都下线的情况，建议先下线从节点再下线主节点，防止不必要的全量复

制。对于 6381 和 6384 节点下线操作，命令如下：

redis-trib.rb del-node 127.0.0.1:6379

4fa7eac4080f0b667ffeab9b87841da49b84a6e4 # 从节点 6384 id

redis-trib.rb del-node 127.0.0.1:6379

40b8d09d44294d2e23c7c768efc8fcd153446746 #主节点 6381 id

节点下线后确认节点状态：

127.0.0.1:6379> cluster nodes

集群节点状态中已经不包含 6384 和 6381 节点，到目前为止，我们完成了节点的安全下线，新的集群结构如下图所示：

10.5 请求路由

目前我们已经搭建好 Redis 集群并且理解了通信和伸缩细节，但还没有使用客户端去操作集群。Redis 集群对客户端通信协议做了比较大的修改，为了追求性能最大化，并没有采用代理的方式而是采用客户端直连节点的方式。因此对于希望从单机切换到集群环境的应用需要修改客户端代码。

2.Smart 客户端——JedisCluster

（1）JedisCluster 的定义

Jedis 为 Redis Cluster 提供了 Smart 客户端，对应的类是 JedisCluster，它的初始化方法

如下：

public JedisCluster(Set<HostAndPort> jedisClusterNode, int connectionTimeout, int 
 soTimeout, int maxAttempts, final GenericObjectPoolConfig poolConfig) {
 ...
}

其中包含了 5 个参数：

·Set<HostAndPort>jedisClusterNode：所有 Redis Cluster 节点信息（也可以是一部分，因为客户端可以通过 cluster slots 自动发现）。

·int connectionTimeout：连接超时。

·int soTimeout：读写超时。

·int maxAttempts：重试次数。

·GenericObjectPoolConfig poolConfig：连接池参数，JedisCluster 会为 Redis Cluster 的每个节点创建连接池。

例如下面代码展示了一次 JedisCluster 的初始化过程

1. // 初始化所有节点(例如 6 个节点)
2. Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>();
3. jedisClusterNode.add(new HostAndPort("10.10.xx.1", 6379));
4. jedisClusterNode.add(new HostAndPort("10.10.xx.2", 6379));
5. jedisClusterNode.add(new HostAndPort("10.10.xx.3", 6379));
6. jedisClusterNode.add(new HostAndPort("10.10.xx.4", 6379));
7. jedisClusterNode.add(new HostAndPort("10.10.xx.5", 6379));
8. jedisClusterNode.add(new HostAndPort("10.10.xx.6", 6379));
9. // 初始化 commnon-pool 连接池, 并设置相关参数
10. GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
11. // 初始化 JedisCluster
12. JedisCluster jedisCluster = new JedisCluster(jedisClusterNode, 1000,
1000, 5, poolConfig);

JedisCluster 可以实现命令的调用，如下所示。

jedisCluster.set("hello", "world");
jedisCluster.get("key");

（2）多节点命令和操作

Redis Cluster 虽然提供了分布式的特性，但是有些命令或者操作，诸如 keys、

flushall、删除指定模式的键，需要遍历所有节点才可以完成。下面代码实现了从 Redis Cluster 删除指定模式键的功能：

1. // 从 RedisCluster 批量删除指定 pattern 的数据
2. public void delRedisClusterByPattern(JedisCluster jedisCluster, String patte
rn,
3. int scanCounter) {
4. // 获取所有节点的 JedisPool
5. Map<String, JedisPool> jedisPoolMap = jedisCluster.getClusterNodes();
6. for (Entry<String, JedisPool> entry : jedisPoolMap.entrySet()) {
7. // 获取每个节点的 Jedis 连接
8. Jedis jedis = entry.getValue().getResource();
9. // 只删除主节点数据
10. if (!isMaster(jedis)) {
11. continue;
12. }
13. // 使用 Pipeline 每次删除指定前缀的数据
14. Pipeline pipeline = jedis.pipelined();
15. // 使用 scan 扫描指定前缀的数据
16. String cursor = "0";
17. // 指定扫描参数：每次扫描个数和 pattern
18. ScanParams params = new ScanParams().count(scanCounter).match(patter
n);
19. while (true) {
20. // 执行扫描
21. ScanResult<String> scanResult = jedis.scan(cursor, params);
22. // 删除的 key 列表
23. List<String> keyList = scanResult.getResult();
24. if (keyList != null && keyList.size() > 0) {
25. for (String key : keyList) {
26. pipeline.del(key);
27. }
28. // 批量删除
29. pipeline.syncAndReturnAll();
30. }
31. cursor = scanResult.getStringCursor();
32. // 如果游标变为 0, 说明扫描完毕
33. if ("0".equals(cursor)) {更多 IT 认证课程请访问 美
break;
35. }
36. }
37. }
38. }
39. // 判断当前 Redis 是否为 master 节点
40. private boolean isMaster(Jedis jedis) {
41. String[] data = jedis.info("Replication").split("\r\n");
42. for (String line : data) {
43. if ("role:master".equals(line.trim())) {
44. return true;
45. }
46. }
47. return false;
48. }

具体分为如下几个步骤：

1）通过 jedisCluster.getClusterNodes（）获取所有节点的连接池。

2）使用 info replication 筛选 1）中的主节点。

3）遍历主节点，使用 scan 命令找到指定模式的 key，使用 Pipeline 机制删除。

例如下面操作每次遍历 1000 个 key，将 Redis Cluster 中以 user 开头的 key 全

部删除。

String pattern = "user*";

int scanCounter = 1000;

delRedisClusterByPattern(jedisCluster, pattern, scanCounter);

所以对于 keys、flushall 等需要遍历所有节点的命令，同样可以参照上面的方法进行相应功能的实现。

（3）批量操作的方法更多 IT 认证课程请访问美河学习在线 ww

Redis Cluster 中，由于 key 分布到各个节点上，会造成无法实现 mget、mset 等功能。但是可以利用 CRC16 算法计算出 key 对应的 slot，以及 Smart 客户端保存了 slot 和节点对应关系的特性，将属于同一个 Redis 节点的 key 进行归档，然后分别对每个节点对应的子 key 列表执行 mget 或者 pipeline 操作。

（4）使用 Lua、事务等特性的方法

Lua 和事务需要所操作的 key，必须在一个节点上，不过 Redis Cluster 提供了hashtag，如果开发人员确实要使用 Lua 或者事务，可以将所要操作的 key 使用一个 hashtag，如下所示：

1. // hashtag
2. String hastag = "{user}";
3. // 用户 A 的关注表
4. String userAFollowKey = hastag + ":a:follow";
5. // 用户 B 的粉丝表
6. String userBFanKey = hastag + ":b:fans";
7. // 计算 hashtag 对应的 slot
8. int slot = JedisClusterCRC16.getSlot(hastag);
9. // 获取指定 slot 的 JedisPool
10. JedisPool jedisPool = jedisCluster.getConnectionHandler().getJedisPoolFromSl
ot(slot);
11. // 在当个节点上执行事务
12. Jedis jedis = null;
13. try {
14. jedis = jedisPool.getResource();
15. // 用户 A 的关注表加入用户 B, 用户 B 的粉丝列表加入用户 A
16. Transaction transaction = jedis.multi();
17. transaction.sadd(userAFollowKey, "user:b");
18. transaction.sadd(userBFanKey, "user:a");
19. transaction.exec();
20. } catch (Exception e) {
21. logger.error(e.getMessage(), e);
22. } finally {
23. if (jedis!= null)
24. jedis.close();
25. }

将事务中所有的 key 添加 hashtag。

2）使用 CRC16 计算 hashtag 对应的 slot。

3）获取指定 slot 对应的节点连接池 JedisPool。

4）在 JedisPool 上执行事务。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2022-02-10，如有侵权请联系 cloudcommunity@tencent.com 删除

node.js