1. 前言
Amazon ElastiCache 是一种 Web 服务，可让用户在云中轻松设置、管理和扩展分布式内存数据存储或缓存环境。它可以提供高性能、可扩展且具有成本效益的缓存解决方案。同时，它可以帮助消除与部署和管理分布式缓存环境相关的复杂性。
ElastiCache for Redis 集群是一个或多个缓存节点的集合，其中所有节点都运行 Redis 缓存引擎软件的实例。ElastiCache for Redis 启用集群模式比非集群模式拥有更好的可扩展性尤其是写入可扩展性，更强的高可用性以及更高的资源上限，因而现在越来越多的客户选择 ElastiCache for Redis 启用集群模式。相比非集群模式，集群模式在多个方面都展现出显著优势：它支持多达 500个节点，存储容量可达 340 TB，并发连接数可高达 3250 万，故障恢复时间仅需 10-20 秒，支持水平和垂直双向扩展，可以通过增加分片或副本来提升性能和可用性。这种灵活的架构使得 Redis 集群能够适应从小型应用到大规模企业级部署的各种需求，为现代高并发、大数据量的应用提供了一个高性能、高可用性的缓存解决方案。
要使用 ElastiCache for Redis 集群（启用集群模式），您需要使用可以支持 Redis 集群模式的客户端。集群模式采用了 16384 个哈希槽来均匀分配数据，通过 CRC16(key) mod 16384 算法精确定位每个键的位置。这些槽位分布在集群的多个分片上，具有集群感知能力的客户端能够正确将请求重定向到正确的分片，并缓存分片映射关系，大大提高了系统效率。
在当今微服务架构盛行的时代，Spring Boot 是众多企业 Java 开发者的常用框架，集成了 Lettuce 作为默认的 Redis 客户端，这是一个多线程连接安全的客户端；Spring Data Redis 也是 Spring 开发项目中经常使用的组件，提供了对 Redis 的轻松配置和使用。
本博客将展示如何使用 Spring Data Redis 来连接并使用 Amazon Elasticache for Redis 集群模式。
2. Spring Data Redis 配置
2.1 引入 Spring Data Redis 及其依赖，配置 Lettuce 作为 Redis 客户端
众所周知，Spring boot 受欢迎一个很重要原因是拥有众多 spring starter 组件，Spring Data Redis 亦不例外，https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-data-redis 利用 Starter 组件，我们可以快速利用如下配置实现引入 Spring Data Redis 并且采用 Lettuce 作为 RedisTemplate 的客户端。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.1</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.amazon.twu</groupId>
<artifactId>demo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>demo</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>17</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
<version>2.11.1</version>
</dependency>
<dependency>
<groupId>io.lettuce</groupId>
<artifactId>lettuce-core</artifactId>
<version>6.1.5.RELEASE</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.13.3</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
...
...
2.2 采用基于 properties 方式配置 Spring Data Redis 来简化 Spring 代码，配置示例如下所示
spring:
redis:
client-type: lettuce
connect-timeout: 3000
timeout: 5000
cluster:
nodes: ${REDIS-CLUSTER-CFG-ENDPOINT}:${REDIS-PORT}
max-redirects: 3
lettuce:
pool:
max-active: 200
max-idle: 8
min-idle: 0
max-wait: 1000
cluster:
refresh:
period: 10
dynamic-refreshSources: true
adaptive: true
@Configuration
@EnableConfigurationProperties(RedisProperties.class)
public class RedisClusterConfig {
@Bean
RedisTemplate<String,Object> redisTemplate(LettuceConnectionFactory connectionFactory){
RedisTemplate<String,Object> rTemplate = new RedisTemplate<>();
rTemplate.setKeySerializer(new StringRedisSerializer());
rTemplate.setValueSerializer(new StringRedisSerializer());
rTemplate.setConnectionFactory(connectionFactory);
rTemplate.afterPropertiesSet();
return rTemplate;
}
@Bean
public LettuceClientConfigurationBuilderCustomizer lettuceClientConfigurationBuilderCustomizer() {
return p -> p.readFrom(ReadFrom.ANY_REPLICA);
}
}
3. 测试集群准备
创建 ElastiCache for Redis 集群（启用集群模式）的详细步骤参考官方文档。
使用的测试集群配置为三分片，每分片（节点组）包含至少两节点，slots 分布如下图所示：
4. Mget 测试
Lettuce 支持 MGET 指令，以下代码通过列表一次性返回所有传入 Key 的结果：
@Test
void testMGET(){
var keyList = Stream.of("top","dummykey").collect(toList());
var result = rTemplate.opsForValue().multiGet(keyList);
result.forEach(res -> logger.info("mGet result is {}",res));
}
可以看到能够通过 mget 从不同的分片中获取到 top 键和 dummykey 键的值及 slot 信息：
5. Pipeline 测试
以下代码使用 executePipelined 执行管道操作，将 member-1 到 member-100 的键通过 sadd 添加到 demo-set 集合中，并添加断言进行验证：
@Test
void testPipeLine(){
rTemplate.executePipelined(new RedisCallback() {
@Override
@Nullable
public Object doInRedis(RedisConnection connection) throws DataAccessException {
IntStream.rangeClosed(1,100)
.forEach(i -> connection.sAdd("demo-set".getBytes(), String.valueOf("member-"+i).getBytes()));
return null;
}
});
assertEquals(Boolean.TRUE, rTemplate.opsForSet().isMember("demo-set", "member-95"));
assertEquals(Boolean.FALSE, rTemplate.opsForSet().isMember("demo-set", "member-101"));
}
6. 读写分离测试
默认情况下，Lettuce 会使用每个节点组的主节点来读写数据，我们可以定制化配置使 Lettuce 客户端从只读副本节点读数据，从主节点写数据。
相关的配置代码如下所示，需要注意的是 ReadFrom 有多个选项可以使用，此处我们指定从任意只读副本节点读取：
@Bean
public LettuceClientConfigurationBuilderCustomizer lettuceClientConfigurationBuilderCustomizer() {
return p -> p.readFrom(ReadFrom.ANY_REPLICA);
}
}
在 ElastiCache 控制台的集群页面，可以通过 Is Master 指标查看每个节点的角色，值为 1 的即该节点组的主节点：
在客户端设置读写分离后，可以通过 Get Type Commnads 和 Set Type Commands 指标查看读写请求的节点分布情况，下图可以看到读请求已经按照设置在只读副本节点执行：
7. Pub/Sub 消息测试
以下代码实现了一个基于 Redis 的消息发布/订阅系统，监听 chat 主题、接收记录信息，同时每 3s 查询一次 top 键的信息并向 chat 主题发送一条消息：
public class MessageReceiver {
Logger logger = LoggerFactory.getLogger(this.getClass());
public void onMessage(String msg){
logger.info("Meesage received: {}", msg);
}
}
@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
MessageListenerAdapter listenerAdapter) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.addMessageListener(listenerAdapter, new PatternTopic("chat"));
return container;
}
@Bean
MessageListenerAdapter listenerAdapter(MessageReceiver receiver) {
return new MessageListenerAdapter(receiver, "onMessage");
}
@Bean
MessageReceiver msgReceiver(){
return new MessageReceiver();
}
@Scheduled(fixedDelay = 3000, initialDelay = 1000)
@Profile("refresh")
public void taskrun() throws Exception {
logger.info("executing...");
try{
redisTemplate.opsForValue().setIfAbsent("top", "test-value");
logger.info("the given key top is on slot {} and the value is {} on master node {}"
,String.valueOf(redisTemplate.getConnectionFactory().getClusterConnection().clusterGetSlotForKey("top".getBytes()))
,String.valueOf(redisTemplate.opsForValue().get("top"))
,String.valueOf(redisTemplate.getConnectionFactory().getClusterConnection().clusterGetNodeForKey("top".getBytes()))
);
logger.info("the given key top is on slot {} and the value is {} on master node {}"
,String.valueOf(redisTemplate.getConnectionFactory().getClusterConnection().clusterGetSlotForKey("dummykey".getBytes()))
,String.valueOf(redisTemplate.opsForValue().get("dummykey"))
,String.valueOf(redisTemplate.getConnectionFactory().getClusterConnection().clusterGetNodeForKey("dummykey".getBytes()))
);
redisTemplate.convertAndSend("chat", "Greetings!");
}catch (Exception ex){
logger.error("fetch redis value error", ex.getCause());
}
}
可以看到能够成功发布及订阅消息：
8. Primary failover 测试
使用以下代码以 3s 一次的频率持续向集群发送请求：
@Component
public class ScheduledTask {
private static Logger logger = LoggerFactory.getLogger(ScheduledTask.class);
@Autowired
RedisTemplate<String,Object> redisTemplate;
@Scheduled(fixedDelay = 3000, initialDelay = 1000)
@Profile("refresh")
public void taskrun() throws Exception {
logger.info("executing...");
try{
redisTemplate.opsForValue().setIfAbsent("top", "test-value");
logger.info("the given key top is on slot {} and the value is {} on master node {}"
,String.valueOf(redisTemplate.getConnectionFactory().getClusterConnection().clusterGetSlotForKey("top".getBytes()))
,String.valueOf(redisTemplate.opsForValue().get("top"))
,String.valueOf(redisTemplate.getConnectionFactory().getClusterConnection().clusterGetNodeForKey("top".getBytes()))
);
登录 ElastiCache 控制台进行 failover 操作，在 event 输出中观察 failover 过程，可以看到在 19:34:24-19:35:04 期间，集群的节点组 0002 经过 40s 完成了一次 failover：
观察测试代码的输出，实际中断时间在 26s 左右：
9. 小结
本文展示了如何使用 Spring Data Redis 连接和操作 ElastiCache 集群，从这个简单的 demo 中我们验证了 Spring Data Redis 能很好地和 ElastiCache Redis 集群模式支持 MGET、Pipeline、读写分离、Pub/Sub、Failover 等功能。
本篇作者
