Redis从入门到放弃系列(十) Cluster

Redis从入门到放弃系列(十) Cluster

本文例子基于：5.0.4

Redis Cluster集群高可用方案,去中心化,最基本三主多从,主从切换类似Sentinel,关于Sentinel内容可以查看编者另外一篇【Redis从入门到放弃系列(九) Sentinel】.

在Redis Cluster中,只存在index为0的数据库,而且其实Redis作为单线程,如果在同一个实例上创建多个库的话,也是需要上下文切换的.

slot

由于Redis Cluster是采用16384个slot来划分数据的,也就是说你当前插入的数据会存在不同的节点上,简而言之不支持比较复杂的多建操作(可以对key打上hash tags来解决).

我们说Cluster是按照16384个slot来划分数据的,那么是如何来确定一个key落在那个节点上呢?

1//计算slot
2HASH_SLOT = CRC16(key) mod 16384
3

每个节点会拥有一部分的slot,通过上述获取到具体key的slot即知道应该去哪儿找对应的节点啦.可是在网络中,一切都会有不存稳定因素,网络抖动.

当在Cluster中存在网络抖动的时候,当时间过长,有可能产生下线,其实原理跟Sentinel里面讲的很相似,因为都是依赖Gossip协议来实现的.可以通过以下配置来设置确定下线的时间.

1//节点持续timeout的时间,才认定该节点出现故障,需要进行主从切换,
2cluster-node-timeout
3//作为上面timeout的系数来放大时间
4cluster-replica-validity-factor
5

由于数据是按照16384个slot去划分的,那么当我们在请求某个key到错误的节点,这时候key不在该节点上,Redis会向我们发送一个错误

1-MOVED 3999 127.0.0.1:6381
2

该消息是提示我们该key应该是存在127.0.0.1这台服务器上面的3999slot,这时候就需要我们的redis客户端去纠正本地的slot映射表,然后请求对应的地址.

增删集群节点

当我们在增加或者删除某个节点的时候,其实就只是将slot从某个节点移动到另外一个节点.可以使用一下命令来完成这一件事

• CLUSTER ADDSLOTS slot1 [slot2] ... [slotN]
• CLUSTER DELSLOTS slot1 [slot2] ... [slotN]
• CLUSTER SETSLOT slot NODE node
• CLUSTER SETSLOT slot MIGRATING node
• CLUSTER SETSLOT slot IMPORTING node 有时候运维需要对redis节点的某些数据做迁移,官方提供了redis-trib工具来完成这件事情。

在迁移的时候,redis节点会存在两种状态,一种是MIGRATING和IMPORTING,用于将slot从一个节点迁移到另外一个节点.

• 节点状态设置为MIGRATING时，将接受与此散列槽有关的所有查询，但仅当有问题的key存在时才能接受，否则将使用-Ask重定向将查询转发到作为迁移目标的节点。
• 节点状态设置为IMPORTING时，节点将接受与此哈希槽有关的所有查询，但前提是请求前面有ASKING命令。如果客户端没有发出ASKING命令，查询将通过-MOVED重定向错误重定向到真正的散列槽所有者

多线程批量获取/删除

1public class RedisUtils {
2
3	private static final String LOCK_SUCCESS = "OK";
4	private static final String SET_IF_NOT_EXIST = "NX";
5	private static final String SET_WITH_EXPIRE_TIME = "PX";
6	private static final Long RELEASE_SUCCESS = 1L;
7
8	private final ThreadLocal<String> requestId = new ThreadLocal<>();
9
10	private final static ExecutorService executorService = new ThreadPoolExecutor(
11			//核心线程数量
12			1,
13			//最大线程数量
14			8,
15			//当线程空闲时，保持活跃的时间
16			1000,
17			//时间单元 ，毫秒级
18			TimeUnit.MILLISECONDS,
19			//线程任务队列
20			new LinkedBlockingQueue<>(1024),
21			//创建线程的工厂
22			new RedisTreadFactory("redis-batch"));
23
24	@Autowired
25	private JedisCluster jedisCluster;
26
27	public String set(String key, String value) {
28		return jedisCluster.set(key, value);
29	}
30
31	public String get(String key) {
32		return jedisCluster.get(key);
33	}
34
35	public Map<String, String> getBatchKey(List<String> keys) {
36		Map<Jedis, List<String>> nodeKeyListMap = jedisKeys(keys);
37		//结果集
38		Map<String, String> resultMap = new HashMap<>();
39		CompletionService<Map<String,String>> batchService = new ExecutorCompletionService(executorService);
40		nodeKeyListMap.forEach((k,v)->{
41			batchService.submit(new BatchGetTask(k,v));
42		});
43		nodeKeyListMap.forEach((k,v)->{
44			try {
45				resultMap.putAll(batchService.take().get());
46			} catch (InterruptedException | ExecutionException e) {
47				e.printStackTrace();
48			}
49		});
50		return resultMap;
51	}
52
53	public boolean lock(String lockKey, long expireTime){
54		String uuid = UUID.randomUUID().toString();
55		requestId.set(uuid);
56		String result = jedisCluster.set(lockKey, uuid, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
57		return LOCK_SUCCESS.equals(result);
58	}
59
60	public boolean unLock(String lockKey){
61		String uuid = requestId.get();
62		String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
63		Object result = jedisCluster.eval(script, Collections.singletonList(lockKey), Collections.singletonList(uuid));
64		requestId.remove();
65		return RELEASE_SUCCESS.equals(result);
66	}
67
68	private Map<Jedis, List<String>> jedisKeys(List<String> keys){
69		Map<Jedis, List<String>> nodeKeyListMap = new HashMap<>();
70		for (String key : keys) {
71			//计算slot
72			int slot = JedisClusterCRC16.getSlot(key);
73			Jedis jedis = jedisCluster.getConnectionFromSlot(slot);
74			if (nodeKeyListMap.containsKey(jedis)) {
75				nodeKeyListMap.get(jedis).add(key);
76			} else {
77				nodeKeyListMap.put(jedis, Arrays.asList(key));
78			}
79		}
80		return nodeKeyListMap;
81	}
82
83	public long delBatchKey(List<String> keys){
84		Map<Jedis, List<String>> nodeKeyListMap = jedisKeys(keys);
85		CompletionService<Long> batchService = new ExecutorCompletionService(executorService);
86		nodeKeyListMap.forEach((k,v)->{
87			batchService.submit(new BatchDelTask(k,v));
88		});
89		Long result = 0L;
90		for (int i=0;i<nodeKeyListMap.size();i++){
91			try {
92				result += batchService.take().get();
93			} catch (InterruptedException | ExecutionException e) {
94				e.printStackTrace();
95			}
96		}
97		return result;
98	}
99
100	class BatchGetTask implements Callable<Map<String,String>>{
101
102		private Jedis jedis;
103
104		private List<String> keys;
105
106		private BatchGetTask(Jedis jedis, List<String> keys) {
107			this.jedis = jedis;
108			this.keys = keys;
109		}
110
111		@Override
112		public Map<String, String> call() throws Exception {
113			Map<String, String> resultMap = new HashMap<>();
114			String[] keyArray = keys.toArray(new String[]{});
115			try {
116				List<String> nodeValueList = jedis.mget(keyArray);
117				for (int i = 0; i < keys.size(); i++) {
118					resultMap.put(keys.get(i),nodeValueList.get(i));
119				}
120			}finally {
121				jedis.close();
122			}
123			return resultMap;
124		}
125	}
126
127	class BatchDelTask implements Callable<Long>{
128
129		private Jedis jedis;
130
131		private List<String> keys;
132
133		private BatchDelTask(Jedis jedis, List<String> keys) {
134			this.jedis = jedis;
135			this.keys = keys;
136		}
137
138		@Override
139		public Long call() throws Exception {
140			String[] keyArray = keys.toArray(new String[]{});
141			try {
142				return jedis.del(keyArray);
143			}finally {
144				jedis.close();
145			}
146		}
147	}
148
149	 static class RedisTreadFactory implements ThreadFactory{
150
151		private final AtomicInteger threadNumber = new AtomicInteger(0);
152
153		private final String namePredix;
154
155		public RedisTreadFactory(String namePredix) {
156			this.namePredix = namePredix +"-";
157		}
158
159		@Override
160		public Thread newThread(Runnable r) {
161			Thread t = new Thread( r,namePredix + threadNumber.getAndIncrement());
162			if (t.isDaemon())
163				t.setDaemon(true);
164			if (t.getPriority() != Thread.NORM_PRIORITY)
165				t.setPriority(Thread.NORM_PRIORITY);
166			return t;
167		}
168	}
169}
170

写在最后

Redis从入门到放弃系列终于完结啦！！！！！！！！！！！

写博客,真的是非常耗时间,真的,本来星期六日要写的,然而因为某些问题而没有写出来(PS:纯粹是因为打游戏.hhhh),终于在今天痛定思痛,顶着脖子酸的压力(PS:贴着狗皮膏药在撸码),终于完结了.

感谢各位看官那么辛苦看我码字,真心感谢.

希望写的东西对各位看官有启发.