cluster info
cluster_state:ok
cluster_slots_assigned:16384
cluster_slots_ok:16384
cluster_slots_pfail:0
cluster_slots_fail:0
cluster_known_nodes:12

xx.xxx.xxx.xx:6959> cluster nodes 45abb8663c0cdb25ed17c29521bf6fda98e913ea xx.xxx.xxx.xx:6961 master - 0 1529229636724 11 connected 13653-15018 e40080f32a3fb89e34b7622038ce490682428fdf
xx.xxx.xxx.xx:6960 master - 0 1529229633723 10 connected 12288-13652 a749bba5614680dea9f47e3c8fe595aa8be71a2c
xx.xxx.xxx.xx:6954 master - 0 1529229639230 4 connected 4096-5460 1096e2a8737401b66c7d4ee0addcb10d7ff14088
xx.xxx.xxx.xx:6952 master - 0 1529229636224 2 connected 1365-2730 fbc76f3481271241c1a89fabeb5139905e1ec2a6
xx.xxx.xxx.xx:6962 master - 0 1529229638230 12 connected 15019-16383 85601fa67820a5af0de0cc21d102d72575709ec6
xx.xxx.xxx.xx:6959 myself,master - 0 0 9 connected 10923-12287 c00d86999c98f97d697f3a2b33ba26fbf50e46eb
xx.xxx.xxx.xx:6955 master - 0 1529229634724 5 connected 5461-6826 0b09a5c4c9e9158520389dd2672bd711d55085c6
xx.xxx.xxx.xx:6953 master - 0 1529229637227 3 connected 2731-4095 9f26d208fa8772449d5c322eb63786a1cf9937e0
xx.xxx.xxx.xx:6958 master - 0 1529229635224 8 connected 9557-10922 274294a88758fcb674e1a0292db0e36a66a0bf48
xx.xxx.xxx.xx:6951 master - 0 1529229634223 1 connected 0-1364 369780bdf56d483a0f0a92cb2baab786844051f3
xx.xxx.xxx.xx:6957 master - 0 1529229640232 7 connected 8192-9556 71ed0215356c664cc56d4579684e86a83dba3a92
xx.xxx.xxx.xx:6956 master - 0 1529229635724 6 connected 6827-8191

redis 127.0.0.1:6379> CLIENT LIST
addr=127.0.0.1:43143 fd=6 age=183 idle=0 flags=N db=0 sub=0 psub=0 multi=-1 qbuf=0 qbuf-free=32768 obl=0 oll=0 omem=0 events=r cmd=client
addr=127.0.0.1:43163 fd=5 age=35 idle=15 flags=N db=0 sub=0 psub=0 multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 events=r cmd=ping
addr=127.0.0.1:43167 fd=7 age=24 idle=6 flags=N db=0 sub=0 psub=0 multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 events=r cmd=get

redis 127.0.0.1:6379> cluster slots
1) 1) (integer) 02) (integer) 40953) 1) "127.0.0.1"2) (integer) 70004) 1) "127.0.0.1"2) (integer) 7004
2) 1) (integer) 122882) (integer) 163833) 1) "127.0.0.1"2) (integer) 70034) 1) "127.0.0.1"2) (integer) 7007
3) 1) (integer) 40962) (integer) 81913) 1) "127.0.0.1"2) (integer) 70014) 1) "127.0.0.1"2) (integer) 7005
4) 1) (integer) 81922) (integer) 122873) 1) "127.0.0.1"2) (integer) 70024) 1) "127.0.0.1"2) (integer) 7006

cluster keyslot test
(integer) 6918

xx.xxx.xxx.xx:6959> get test
(error) MOVED 6918 xx.xxx.xx.xxx:6956　　

xx.xxx.xx.xxx:6956> get test
"cluster"

mget test test1
(error) CROSSSLOT Keys in request don't hash to the same slot

    @Bean(name="clusterRedisURI")RedisURI clusterRedisURI(){return RedisURI.builder().withHost("xx.xx.xxx.xx").withPort(6954).build();}//配置集群选项,自动重连,最多重定型1次@BeanClusterClientOptions clusterClientOptions(){return ClusterClientOptions.builder().autoReconnect(true).maxRedirects(1).build();}//创建集群客户端@BeanRedisClusterClient redisClusterClient(ClientResources clientResources, ClusterClientOptions clusterClientOptions, RedisURI clusterRedisURI){RedisClusterClient redisClusterClient= RedisClusterClient.create(clientResources,clusterRedisURI);redisClusterClient.setOptions(clusterClientOptions);return redisClusterClient;}/*** 集群连接*/@Bean(destroyMethod = "close")StatefulRedisClusterConnection<String,String> statefulRedisClusterConnection(RedisClusterClient redisClusterClient){return  redisClusterClient.connect();}

 public StatefulRedisClusterConnection<String, String> connect() {return connect(newStringStringCodec());}

  <K, V> StatefulRedisClusterConnectionImpl<K, V> connectClusterImpl(RedisCodec<K, V> codec) {//如果分区信息为null则初始化分区信息if (partitions == null) {initializePartitions();}//如果需要就激活拓扑刷新activateTopologyRefreshIfNeeded();

 protected void initializePartitions() {this.partitions = loadPartitions();}

  protected Partitions loadPartitions() {//获取拓扑刷新信息，Iterable<RedisURI> topologyRefreshSource = getTopologyRefreshSource();String message = "Cannot retrieve initial cluster partitions from initial URIs " + topologyRefreshSource;try {//加载拓扑信息Map<RedisURI, Partitions> partitions = refresh.loadViews(topologyRefreshSource, useDynamicRefreshSources());

protected Iterable<RedisURI> getTopologyRefreshSource() {//是否初始化种子节点boolean initialSeedNodes = !useDynamicRefreshSources();Iterable<RedisURI> seed;//如果需要初始化种子节点或分区信息为null或分区信息为空 则将初始URI赋值给种子if (initialSeedNodes || partitions == null || partitions.isEmpty()) {seed = RedisClusterClient.this.initialUris;} else {//不需要初始化种子节点List<RedisURI> uris = new ArrayList<>();for (RedisClusterNode partition : TopologyComparators.sortByUri(partitions)) {uris.add(partition.getUri());}seed = uris;}return seed;}

 protected RedisClusterClient(ClientResources clientResources, Iterable<RedisURI> redisURIs) {super(clientResources);assertNotEmpty(redisURIs);assertSameOptions(redisURIs);//初始化节点this.initialUris = Collections.unmodifiableList(LettuceLists.newList(redisURIs));//根据第一个URI的超时时间作为默认超时时间setDefaultTimeout(getFirstUri().getTimeout());setOptions(ClusterClientOptions.builder().build());}

 protected boolean useDynamicRefreshSources() {//如果集群客户端选项不为nullif (getClusterClientOptions() != null) {//获取集群拓扑刷新选项ClusterTopologyRefreshOptions topologyRefreshOptions = getClusterClientOptions().getTopologyRefreshOptions();//返回集群拓扑刷新选项中配置到是否使用动态刷新return topologyRefreshOptions.useDynamicRefreshSources();}//默认动态刷新return true;}

 public Map<RedisURI, Partitions> loadViews(Iterable<RedisURI> seed, boolean discovery) {//获取超时时间,默认60秒long commandTimeoutNs = getCommandTimeoutNs(seed);Connections connections = null;try {//获取所有种子连接connections = getConnections(seed).get(commandTimeoutNs, TimeUnit.NANOSECONDS);//Requests将异步执行命令封装到多个节点//cluster nodesRequests requestedTopology = connections.requestTopology();//client listRequests requestedClients = connections.requestClients();//获取节点拓扑视图NodeTopologyViews nodeSpecificViews = getNodeSpecificViews(requestedTopology, requestedClients, commandTimeoutNs);if (discovery) {//是否查找额外节点//获取集群节点Set<RedisURI> allKnownUris = nodeSpecificViews.getClusterNodes();//排除种子节点，得到需要发现节点Set<RedisURI> discoveredNodes = difference(allKnownUris, toSet(seed));//如果需要发现节点不为空if (!discoveredNodes.isEmpty()) {//需要发现节点连接Connections discoveredConnections = getConnections(discoveredNodes).optionalGet(commandTimeoutNs,TimeUnit.NANOSECONDS);//合并连接connections = connections.mergeWith(discoveredConnections);//合并请求requestedTopology = requestedTopology.mergeWith(discoveredConnections.requestTopology());requestedClients = requestedClients.mergeWith(discoveredConnections.requestClients());//获取节点视图nodeSpecificViews = getNodeSpecificViews(requestedTopology, requestedClients, commandTimeoutNs);//返回uri对应分区信息return nodeSpecificViews.toMap();}}return nodeSpecificViews.toMap();} catch (InterruptedException e) {Thread.currentThread().interrupt();throw new RedisCommandInterruptedException(e);} finally {if (connections != null) {connections.close();}}}

class ClusterDistributionChannelWriter implements RedisChannelWriter {//默认写入器private final RedisChannelWriter defaultWriter;//集群事件监听器private final ClusterEventListener clusterEventListener;private final int executionLimit;//集群连接提供器private ClusterConnectionProvider clusterConnectionProvider;//异步集群连接提供器private AsyncClusterConnectionProvider asyncClusterConnectionProvider;//是否关闭private boolean closed = false;//分区信息private volatile Partitions partitions;

public <K, V, T> RedisCommand<K, V, T> write(RedisCommand<K, V, T> command) {LettuceAssert.notNull(command, "Command must not be null");//如果连接已经关闭则抛出异常if (closed) {throw new RedisException("Connection is closed");}//如果是集群命令且命令没有处理完毕if (command instanceof ClusterCommand && !command.isDone()) {//类型转换， 转换为ClusterCommandClusterCommand<K, V, T> clusterCommand = (ClusterCommand<K, V, T>) command;if (clusterCommand.isMoved() || clusterCommand.isAsk()) {HostAndPort target;boolean asking;//如果集群命令已经迁移,此时通过ClusterCommand中到重试操作进行到此if (clusterCommand.isMoved()) {//获取命令迁移目标节点target = getMoveTarget(clusterCommand.getError());//触发迁移事件clusterEventListener.onMovedRedirection();asking = false;} else {//如果是asktarget = getAskTarget(clusterCommand.getError());asking = true;clusterEventListener.onAskRedirection();}command.getOutput().setError((String) null);//连接迁移后的目标节点CompletableFuture<StatefulRedisConnection<K, V>> connectFuture = asyncClusterConnectionProvider.getConnectionAsync(ClusterConnectionProvider.Intent.WRITE, target.getHostText(), target.getPort());//成功建立连接,则向该节点发送命令if (isSuccessfullyCompleted(connectFuture)) {writeCommand(command, asking, connectFuture.join(), null);} else {connectFuture.whenComplete((connection, throwable) -> writeCommand(command, asking, connection, throwable));}return command;}}//不是集群命令就是RedisCommand，第一个请求命令就是非ClusterCommand//将当前命令包装为集群命令ClusterCommand<K, V, T> commandToSend = getCommandToSend(command);//获取命令参数CommandArgs<K, V> args = command.getArgs();//排除集群路由的cluster命令if (args != null && !CommandType.CLIENT.equals(commandToSend.getType())) {//获取第一个编码后的keyByteBuffer encodedKey = args.getFirstEncodedKey();//如果encodedKey不为nullif (encodedKey != null) {//获取slot值int hash = getSlot(encodedKey);//根据命令类型获取命令意图 是读还是写ClusterConnectionProvider.Intent intent = getIntent(command.getType());//根据意图和slot获取连接CompletableFuture<StatefulRedisConnection<K, V>> connectFuture = ((AsyncClusterConnectionProvider) clusterConnectionProvider).getConnectionAsync(intent, hash);//如果成功获取连接if (isSuccessfullyCompleted(connectFuture)) {writeCommand(commandToSend, false, connectFuture.join(), null);} else {//如果连接尚未处理完，或有异常，则添加完成处理器connectFuture.whenComplete((connection, throwable) -> writeCommand(commandToSend, false, connection,throwable));}return commandToSend;}}writeCommand(commandToSend, defaultWriter);return commandToSend;}

 @Overridepublic void complete() {//如果响应是MOVED或ASKif (isMoved() || isAsk()) {//如果最大重定向次数大于当前重定向次数则可以进行重定向boolean retryCommand = maxRedirections > redirections;//重定向次数自增redirections++;if (retryCommand) {try {//重定向retry.write(this);} catch (Exception e) {completeExceptionally(e);}return;}}super.complete();completed = true;}

 private static <K, V> void writeCommand(RedisCommand<K, V, ?> command, boolean asking,StatefulRedisConnection<K, V> connection, Throwable throwable) {if (throwable != null) {command.completeExceptionally(throwable);return;}try {//如果需要发送asking请求,即接收到ASK错误消息,则在重定向到目标主机后需要发送asking命令if (asking) {connection.async().asking();}//发送命令writeCommand(command, ((RedisChannelHandler<K, V>) connection).getChannelWriter());} catch (Exception e) {command.completeExceptionally(e);}}

  @Overridepublic RedisFuture<List<KeyValue<K, V>>> mget(Iterable<K> keys) {//获取分区和key的映射关系Map<Integer, List<K>> partitioned = SlotHash.partition(codec, keys);//如果分区数小于2也就是只有一个分区即所有key都落在一个分区就直接获取if (partitioned.size() < 2) {return super.mget(keys);}//每个key与slot映射关系Map<K, Integer> slots = SlotHash.getSlots(partitioned);Map<Integer, RedisFuture<List<KeyValue<K, V>>>> executions = new HashMap<>();//遍历分片信息，逐个发送for (Map.Entry<Integer, List<K>> entry : partitioned.entrySet()) {RedisFuture<List<KeyValue<K, V>>> mget = super.mget(entry.getValue());executions.put(entry.getKey(), mget);}//恢复key的顺序return new PipelinedRedisFuture<>(executions, objectPipelinedRedisFuture -> {List<KeyValue<K, V>> result = new ArrayList<>();for (K opKey : keys) {int slot = slots.get(opKey);int position = partitioned.get(slot).indexOf(opKey);RedisFuture<List<KeyValue<K, V>>> listRedisFuture = executions.get(slot);result.add(MultiNodeExecution.execute(() -> listRedisFuture.get().get(position)));}return result;});}