@Configuration
public class LettuceConfig {/*** 配置客户端资源* @return*/@Bean(destroyMethod = "shutdown")ClientResources clientResources() {return DefaultClientResources.builder().ioThreadPoolSize(8).computationThreadPoolSize(10).build();}/*** 配置Socket选项* keepAlive=true* tcpNoDelay=true* connectionTimeout=5秒* @return*/@BeanSocketOptions socketOptions(){return SocketOptions.builder().keepAlive(true).tcpNoDelay(true).connectTimeout(Duration.ofSeconds(5)).build(); }/*** 配置客户端选项* @return*/@BeanClientOptions clientOptions(SocketOptions socketOptions) {return ClientOptions.builder().socketOptions(socketOptions).build();}/*** 创建RedisClient* @param clientResources 客户端资源* @param clientOptions 客户端选项* @return */@Bean(destroyMethod = "shutdown")RedisClient redisClient(ClientResources clientResources, ClientOptions clientOptions) {RedisURI uri = RedisURI.builder().withSentinel("xx.xx.xx.xx", 26009).withPassword("abcd1234").withSentinelMasterId("xxx").build();RedisClient client = RedisClient.create(clientResources, uri);client.setOptions(clientOptions);return client;}/*** 创建连接* @param redisClient* @return*/@Bean(destroyMethod = "close")StatefulRedisConnection<String, String> connection(RedisClient redisClient) {return redisClient.connect();}
}

public Mono<ServerResponse> hello(ServerRequest request) throws  Exception {//响应式使用Mono<String> resp = redisConnection.reactive().get("gxt_new");//同步使用redisConnection.sync().get("test");redisConnection.async().get("test").get(5, TimeUnit.SECONDS);return ServerResponse.ok().body(resp, String.class);
}

client.getResources().eventBus().get().subscribe(e -> {System.out.println("client 订阅事件： " + e);});

client 订阅事件： ConnectionActivatedEvent [/xx:49910 -> /xx:6008]
client 订阅事件： ConnectionActivatedEvent [/xx:49911 -> /xx:6018]
client 订阅事件： ConnectedEvent [/xx:49912 -> /xx:6018]

 eventBus.publish(new Event() {@Overridepublic String toString() {return "自定义事件";}});

client 订阅事件： 自定义事件

@Bean(destroyMethod = "close")StatefulRedisMasterSlaveConnection<String, String> statefulRedisMasterSlaveConnection(RedisClient redisClient, RedisURI redisURI) {StatefulRedisMasterSlaveConnection connection = MasterSlave.connect(redisClient, new Utf8StringCodec(), redisURI);connection.setReadFrom(ReadFrom.NEAREST);return connection;}
}

 //根据意图获取连接public StatefulRedisConnection<K, V> getConnection(Intent intent) {if (debugEnabled) {logger.debug("getConnection(" + intent + ")");}//如果readFrom不为null且是READif (readFrom != null && intent == Intent.READ) {//根据readFrom配置从已知节点中选择可用节点描述List<RedisNodeDescription> selection = readFrom.select(new ReadFrom.Nodes() {@Overridepublic List<RedisNodeDescription> getNodes() {return knownNodes;}@Overridepublic Iterator<RedisNodeDescription> iterator() {return knownNodes.iterator();}});//如果可选择节点集合为空则抛出异常if (selection.isEmpty()) {throw new RedisException(String.format("Cannot determine a node to read (Known nodes: %s) with setting %s",knownNodes, readFrom));}try {//遍历所有可用节点for (RedisNodeDescription redisNodeDescription : selection) {//获取节点连接StatefulRedisConnection<K, V> readerCandidate = getConnection(redisNodeDescription);//如果节点连接不是打开到连接则继续查找下一个连接if (!readerCandidate.isOpen()) {continue;}//返回可用连接return readerCandidate;}//如果没有找到可用连接，默认返回第一个return getConnection(selection.get(0));} catch (RuntimeException e) {throw new RedisException(e);}}//如果没有配置readFrom或者不是READ 则返回master连接return getConnection(getMaster());}

static final class ReadFromSlavePreferred extends ReadFrom {@Overridepublic List<RedisNodeDescription> select(Nodes nodes) {List<RedisNodeDescription> result = new ArrayList<>(nodes.getNodes().size());//优先添加slave节点for (RedisNodeDescription node : nodes) {if (node.getRole() == RedisInstance.Role.SLAVE) {result.add(node);}}//最后添加master节点for (RedisNodeDescription node : nodes) {if (node.getRole() == RedisInstance.Role.MASTER) {result.add(node);}}return result;}

 static final class ReadFromMasterPreferred extends ReadFrom {@Overridepublic List<RedisNodeDescription> select(Nodes nodes) {List<RedisNodeDescription> result = new ArrayList<>(nodes.getNodes().size());//优先添加master节点for (RedisNodeDescription node : nodes) {if (node.getRole() == RedisInstance.Role.MASTER) {result.add(node);}}//其次在添加slave节点for (RedisNodeDescription node : nodes) {if (node.getRole() == RedisInstance.Role.SLAVE) {result.add(node);}}return result;}}

 static final class ReadFromSlave extends ReadFrom {@Overridepublic List<RedisNodeDescription> select(Nodes nodes) {List<RedisNodeDescription> result = new ArrayList<>(nodes.getNodes().size());//只获取slave节点for (RedisNodeDescription node : nodes) {if (node.getRole() == RedisInstance.Role.SLAVE) {result.add(node);}}return result;}}

static final class ReadFromMaster extends ReadFrom {@Overridepublic List<RedisNodeDescription> select(Nodes nodes) {for (RedisNodeDescription node : nodes) {if (node.getRole() == RedisInstance.Role.MASTER) {return LettuceLists.newList(node);}}return Collections.emptyList();}}

static final class ReadFromNearest extends ReadFrom {@Overridepublic List<RedisNodeDescription> select(Nodes nodes) {return nodes.getNodes();}}

public List<RedisNodeDescription> getNodes() {logger.debug("lookup topology for masterId {}", masterId);try (StatefulRedisSentinelConnection<String, String> connection = redisClient.connectSentinel(CODEC, sentinelUri)) {RedisFuture<Map<String, String>> masterFuture = connection.async().master(masterId);RedisFuture<List<Map<String, String>>> slavesFuture = connection.async().slaves(masterId);List<RedisNodeDescription> result = new ArrayList<>();try {Map<String, String> master = masterFuture.get(timeout.toNanos(), TimeUnit.NANOSECONDS);List<Map<String, String>> slaves = slavesFuture.get(timeout.toNanos(), TimeUnit.NANOSECONDS);//添加master节点    result.add(toNode(master, RedisInstance.Role.MASTER));//添加所有slave节点result.addAll(slaves.stream().filter(SentinelTopologyProvider::isAvailable).map(map -> toNode(map, RedisInstance.Role.SLAVE)).collect(Collectors.toList()));} catch (ExecutionException | InterruptedException | TimeoutException e) {throw new RedisException(e);}return result;}}

 @Bean(destroyMethod = "close")StatefulRedisMasterSlaveConnection<String, String> statefulRedisMasterSlaveConnection(RedisClient redisClient, RedisURI redisURI) {StatefulRedisMasterSlaveConnection connection = MasterSlave.connect(redisClient, new Utf8StringCodec(), redisURI);connection.setReadFrom(new ReadFrom() {@Overridepublic List<RedisNodeDescription> select(Nodes nodes) {List<RedisNodeDescription> list = nodes.getNodes();Collections.shuffle(list);return list;}});return connection;}

public class Sharded< C extends StatefulRedisConnection,V> {private TreeMap<Long, String> nodes;private final Hashing algo = Hashing.MURMUR_HASH;private final Map<String, StatefulRedisConnection> resources = new LinkedHashMap<>();private RedisClient redisClient;private String password;private Set<HostAndPort> sentinels;private RedisCodec<String, V> codec;public Sharded(List<String> masters, RedisClient redisClient, String password, Set<HostAndPort> sentinels, RedisCodec<String, V> codec) {this.redisClient = redisClient;this.password = password;this.sentinels = sentinels;this.codec = codec;initialize(masters);}private void initialize(List<String> masters) {nodes = new TreeMap<>();for (int i = 0; i != masters.size(); ++i) {final String master = masters.get(i);for (int n = 0; n < 160; n++) {nodes.put(this.algo.hash("SHARD-" + i + "-NODE-" + n), master);}RedisURI.Builder builder = RedisURI.builder();for (HostAndPort hostAndPort : sentinels) {builder.withSentinel(hostAndPort.getHostText(), hostAndPort.getPort());}RedisURI redisURI = builder.withPassword(password).withSentinelMasterId(master).build();resources.put(master, MasterSlave.connect(redisClient, codec, redisURI));}}public StatefulRedisConnection getConnectionBy(String key) {return resources.get(getShardInfo(SafeEncoder.encode(key)));}public Collection<StatefulRedisConnection> getAllConnection(){return Collections.unmodifiableCollection(resources.values());}public String getShardInfo(byte[] key) {SortedMap<Long, String> tail = nodes.tailMap(algo.hash(key));if (tail.isEmpty()) {return nodes.get(nodes.firstKey());}return tail.get(tail.firstKey());}public void close(){for(StatefulRedisConnection connection:  getAllConnection()){connection.close();}}private static  class SafeEncoder {static byte[] encode(final String str) {try {if (str == null) {throw new IllegalArgumentException("value sent to redis cannot be null");}return str.getBytes("UTF-8");} catch (UnsupportedEncodingException e) {throw new RuntimeException(e);}}}private interface Hashing {Hashing MURMUR_HASH = new MurmurHash();long hash(String key);long hash(byte[] key);}private static  class MurmurHash implements Hashing {static long hash64A(byte[] data, int seed) {return hash64A(ByteBuffer.wrap(data), seed);}static long hash64A(ByteBuffer buf, int seed) {ByteOrder byteOrder = buf.order();buf.order(ByteOrder.LITTLE_ENDIAN);long m = 0xc6a4a7935bd1e995L;int r = 47;long h = seed ^ (buf.remaining() * m);long k;while (buf.remaining() >= 8) {k = buf.getLong();k *= m;k ^= k >>> r;k *= m;h ^= k;h *= m;}if (buf.remaining() > 0) {ByteBuffer finish = ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN);// for big-endian version, do this first:// finish.position(8-buf.remaining());finish.put(buf).rewind();h ^= finish.getLong();h *= m;}h ^= h >>> r;h *= m;h ^= h >>> r;buf.order(byteOrder);return h;}public long hash(byte[] key) {return hash64A(key, 0x1234ABCD);}public long hash(String key) {return hash(SafeEncoder.encode(key));}}}

 @Bean(destroyMethod = "close")Sharded<StatefulRedisMasterSlaveConnection,String> sharded(RedisClient redisClient) {Set<HostAndPort> hostAndPorts=new HashSet<>();hostAndPorts.add(HostAndPort.parse("1xx:26009"));hostAndPorts.add(HostAndPort.parse("1xx:26009"));return new Sharded<>(Arrays.asList("te009","test68","test67"),redisClient,"password",hostAndPorts, new Utf8StringCodec());}

  //只从slave节点中读取StatefulRedisMasterSlaveConnection redisConnection = (StatefulRedisMasterSlaveConnection) sharded.getConnectionBy("key");//使用异步模式获取缓存值System.out.println(redisConnection.sync().get("key"));