条条大路通罗马 — 使用 Jedis 访问 Amazon ElastiCache for Redis 集群

一、服务介绍

社区版 Redis 支持灵活的部署架构：单机版、主从版、集群版，能够满足不同的业务场景。亚马逊云科技提供两种与 Redis 兼容的完全托管式服务：Amazon MemoryDB for Redis 和 Amazon ElastiCache for Redis。

Amazon MemoryDB for Redis

Amazon MemoryDB for Redis 是与 Redis 兼容、持久的内存数据库服务，可提供超快的性能。它专为使用微服务架构创建的现代化应用程序构建。Amazon MemoryDB for Redis 使用多可用区事务日志跨多个可用区（AZ）持久存储数据，以实现快速故障转移、数据库恢复和节点重启。借助 Amazon MemoryDB for Redis，您的所有数据都存储在内存中，使您能够实现微秒级读取，以及个位数毫秒的写入延迟和高吞吐量。

Amazon ElastiCache for Redis

Amazon ElastiCache for Redis 是一项完全托管式缓存服务，可让您轻松地在云中设置、操作和扩展缓存。借助 Amazon ElastiCache for Redis，您可以通过缓存来自主管理数据库和数据存储的数据，加快应用程序速度，并获得微秒级读写延迟。

通过上面的说明，大家可能已经注意到 Amazon ElastiCache for Redis 和 Amazon MemoryDB for Redis 有些不同的设计，如果有持久化存储的需求，请选择 Amazon MemoryDB for Redis。

亚马逊云科技平台的 Amazon ElastiCache for Redis 服务为客户提供了两种生产级环境的部署架构：

禁用集群模式：始终有一个分片，最多有 5 个只读副本节点；
集群模式：最多有 500 个分片，每个分片中有 1 到 5 个只读副本节点。

#部署架构图

而亚马逊云科技平台的 Amazon MemoryDB for Redis 服务则是只有集群部署方式。

二、迁移场景

常见的迁移场景中，客户应用程序中的客户端有时会使用 Standalone 模式连所有（禁用集群模式&集群模式），所以当客户在未修改客户端就连接到亚马逊云科技平台上的集群模式 Redis 时，就会导致异常。因为亚马逊云科技平台的 Redis 是严格和社区保持一至的，即在使用不同开发语言的客户端连接时要选择正确的使用模式；如果有认证要求，可使用 Redis AUTH 进行身份验证，只有启用了传输中加密（TLS）的 Redis 服务器才支持启用身份验证，这也要求在使用客户端时要开启 TLS，具体可以参考此文档。

针对以上场景，我们归纳总结一下，当客户希望从其它平台迁移到亚马逊云科技平台并使用托管的 Redis 服务时要考虑如下场景并做适当修改：

原来使用 Redis AUTH（用户名和密码）做认证但未使用 TLS 加密。当迁移到亚马逊云科技平台后，要修改客户端代码，打开 TLS 加密。
原来在 Redis 集群上使用多 DB。因为社区同样不支持在集群模式下使用多 DB，亚马逊云科技与社区方式兼容，同样不能支持，所以当在其它云平台上如果有多 DB 的情况要迁移时，请记得一定要提前拆分或在 Key 没有冲突的情况下合并 DB。
原来使用 Redis 集群模式，但是客户端代码使用 standalone 模式连接集群。当迁移到亚马逊云科技平台后，要修改客户端代码，将 standalone 模式改为 cluster。

三、Redis Client SDK

当您的 Java 程序需要连接和使用 Amazon ElastiCache for Redis 集群时，目前比较主流的 SDK 有 3 种：Jedis，Redisson 和 Lettuce。本篇 Blog 将会为您介绍如何使用 Jedis 连接和使用 Amazon ElastiCache for Redis 集群，另外我们也推出了一系列博客，展示了如何使用其他客户端工具对集群进行连接和操作，欢迎大家阅读。

四、功能测试

4.1 创建子网组

参考文档创建一个 ElastiCache for Redis 用的子网组，命令如下：

aws elasticache create-cache-subnet-group \
--cache-subnet-group-name "my-poc-redis-sg" \
--cache-subnet-group-description "my poc redis subnet group" \
--subnet-ids "subnet-0858xxxx" "subnet-02f2xxxx" "subnet-00e4xxxx"

4.2 创建Amazon ElastiCache for Redis集群（非集群模式）

创建一个非集群模式的 ElastiCache for Redis，配置如下：

实例类型 : cache.t3.small（生产环境建议使用 M 或 R 类型实例）
版本 : 7.0.7
TLS & Auth : 开启
1 Replica

aws elasticache create-replication-group \
--replication-group-id "my-poc-non-cluster-redis-7" \
--replication-group-description "my-poc-non-cluster-redis-7" \
--num-node-groups 1 \
--replicas-per-node-group 1 \
--cache-node-type cache.t3.small \
--cache-parameter-group default.redis7 \
--engine redis \
--engine-version 7.0 \
--cache-subnet-group-name "my-poc-redis-sg" \
--automatic-failover-enabled \
--multi-az-enabled \
--transit-encryption-enabled \
--security-group-ids "sg-04d2xxxx" \
--auth-token yourpassword

最终获取到 Primary endpoint，下文我们会使用到它。

4.3 创建 Amazon ElastiCache for Redis 集群（集群模式）

创建一个集群模式的 ElastiCache for Redis，配置如下：

实例类型 : cache.t3.small
版本 : 7.0.7
TLS & Auth : 开启
一共 3 个 Shard，每个 Shard 有一个 Replica

aws elasticache create-replication-group \
--replication-group-id "my-poc-cluster-redis-7" \
--replication-group-description "my-poc-cluster-redis-7" \
--num-node-groups 3 \
--replicas-per-node-group 1 \
--cache-node-type cache.t3.small \
--cache-parameter-group default.redis7.cluster.on \
--engine redis \
--engine-version 7.0 \
--cache-subnet-group-name "my-poc-redis-sg" \
--automatic-failover-enabled \
--multi-az-enabled \
--transit-encryption-enabled \
--security-group-ids "sg-04d2xxxx" \
--auth-token yourpassword

最终获取到 Configuration endpoint，下文我们会使用到它。

4.4 创建 Cloud9

在同一个 VPC 内创建一个 EC2 并配置好相应的 Security Group 使得 EC2 上的程序可以访问 ElastiCache 集群的端口，在 EC2 上准备好 java 环境。

aws cloud9 create-environment-ec2 \
--name "my-poc-ide" \
--description "my-poc-ide" \
--instance-type "t3.small" \
--subnet-id "subnet-0858xxxx" \
--automatic-stop-time-minutes 60 \
--owner-arn arn:aws:iam::accountid:user/youriamuser

mvn -version
Apache Maven 3.0.5 (Red Hat 3.0.5-17)
Maven home: /usr/share/maven
Java version: 17.0.7, vendor: Amazon.com Inc.
Java home: /usr/lib/jvm/java-17-amazon-corretto.x86_64
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "5.10.177-158.645.amzn2.x86_64", arch: "amd64", family: "unix"

用 maven 创建一个 Java project 并配置 pom.xml

mvn archetype:generate -DgroupId=com.mycompany.app -DartifactId=myapp -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false
cd myapp
vim pom.xml

添加以下红色部分内容，这里我们使用了 Jedis 4.4.3，是笔者在撰写本篇 Blog 时最新的稳定版本，读者可以根据这个链接来找到 Jedis 其他版本。

<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 http://maven.apache.org/maven-v4_0_0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.mycompany.app</groupId>
  <artifactId>myapp</artifactId>
  <packaging>jar</packaging>
  <version>1.0-SNAPSHOT</version>
  <name>myapp</name>
  <url>http://maven.apache.org</url>
  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>redis.clients</groupId>
      <artifactId>jedis</artifactId>
      <version>4.4.3</version>
    </dependency>
  </dependencies>
</project>

接下来我们会介绍 3 种使用 Jedis 连接 Redis 的方式，分别是单机模式，连接池模式和集群模式。

4.4.1 单机模式测试 Code

创建名为 MyJedis.java 的文件，复制并粘贴以下内容

package com.mycompany.app;
import redis.clients.jedis.Jedis;

public class MyJedis
{
    public static void main( String[] args )
    {
        try {
            String host = "Primary endpoint";
            int port = 6379;
            boolean ssl = true;
            String password = "yourpassword";
            Jedis jedis = new Jedis(host, port, ssl);
            jedis.auth(password);
            
            jedis.set("my-poc-key-01", "my-poc-value-01");
            System.out.println(jedis.exists("my-poc-key-01"));
            System.out.println(jedis.get("my-poc-key-01"));
            
            jedis.disconnect();
            jedis.close();
        }catch (Exception e) {
            e.printStackTrace();
        }
        
    }
}

4.4.2 连接池模式测试 Code

创建名为 MyJedisPool.java 的文件，复制并粘贴以下内容

package com.mycompany.app;
import java.time.Duration;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class MyJedisPool 
{
    public static void main( String[] args )
    {
        try {
            String host = "Primary endpoint";
            int port = 6379;
            boolean ssl = true;
            String password = "yourpassword";
            int timeout = 5000;
            
            JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
            jedisPoolConfig.setMaxTotal(1000); //最大活跃连接数
            jedisPoolConfig.setMaxIdle(10);//最大空闲连接数
            jedisPoolConfig.setMinIdle(10);//最小空闲连接数
            jedisPoolConfig.setMaxWait(Duration.ofMillis(5000));//最大的等待时间，如果超过等待时间，则直接抛出JedisConnectionException
        
            JedisPool jedisPool = new JedisPool(
                jedisPoolConfig,
                host,
                port,
                timeout,
                password,
                ssl
            );
        
            Jedis jedisClient = jedisPool.getResource();
            jedisClient.set("my-poc-key-02", "my-poc-value-02");
            System.out.println(jedisClient.exists("my-poc-key-02"));
            System.out.println(jedisClient.get("my-poc-key-02"));
            
            jedisClient.close();
            jedisPool.close();
        }catch (Exception e) {
            e.printStackTrace();
        }
    }
}

4.4.3 集群模式测试 Code

创建名为 MyJedisCluster.java 的文件，复制并粘贴以下内容

package com.mycompany.app;
import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.Connection;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;

public class MyJedisCluster
{
    public static void main( String[] args )
    {
        try {
            String host = "Configuration endpoint";
            int port = 6379;
            int connectionTimeout = 10000;
            int soTimeout = 10000;//读写超时时间
            int maxAttempts = 20;//最大重试次数
            String password = "yourpassword";
            
            Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
            jedisClusterNodes.add(new HostAndPort(host, port));

            GenericObjectPoolConfig<Connection> jedisPoolConfig = new GenericObjectPoolConfig<Connection>();
            jedisPoolConfig.setMaxTotal(100);
            jedisPoolConfig.setMaxIdle(10);
            jedisPoolConfig.setMinIdle(10);
            jedisPoolConfig.setMaxWait(Duration.ofMillis(5000));
            
            JedisCluster jedisCluster = new JedisCluster(
                jedisClusterNodes, 
                connectionTimeout, 
                soTimeout,
                maxAttempts,
                password,
                "mypoc",
                jedisPoolConfig,
                true
            );
            jedisCluster.set("my-poc-key-03", "my-poc-value-03");
            System.out.println(jedisCluster.exists("my-poc-key-03"));
            System.out.println(jedisCluster.get("my-poc-key-03"));
            jedisCluster.close();
            
        }catch (Exception e) {
            e.printStackTrace();
        }
        
    }
}

4.5 运行测试代码

4.5.1 测试单机模式

运行结果应该如下，表明单机模式下 Jedis 已成功 set 了“my-poc-key-01″。

true
my-poc-value-01

4.5.2 测试连接池模式

运行结果应该如下，表明连接池模式下 Jedis 已成功 set 了“my-poc-key-02″。

true
my-poc-value-02

4.5.3 测试集群模式

运行结果应该如下，表明集群模式下 Jedis 已成功 set 了“my-poc-key-03″。

true
my-poc-value-03

五、分片

当 Amazon ElastiCache for Redis 使用集群模式并且为每个 Shard 配置 Replica 时，Jedis 默认是支持分片功能的。下来让我们简单验证一下。

5.1 写入测试 Code

创建名为 MyJedisSharding.java 的文件，复制并粘贴以下内容，我们会使用 Jedis 以集群模式往 Redis 中插入 150 个 Key。

package com.mycompany.app;
import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.Connection;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;

public class MyJedisSharding
{
	public static void main( String[] args )
    {
		try {
			String host = "Configuration endpoint";
			int port = 6379;
			int connectionTimeout = 10000;
            int soTimeout = 10000;
            int maxAttempts = 20;
			String password = "yourpassword";
			
			Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
        	jedisClusterNodes.add(new HostAndPort(host, port));

			GenericObjectPoolConfig<Connection> jedisPoolConfig = new GenericObjectPoolConfig<Connection>();
        	jedisPoolConfig.setMaxTotal(100);
        	jedisPoolConfig.setMaxIdle(10);
        	jedisPoolConfig.setMinIdle(10);
        	jedisPoolConfig.setMaxWait(Duration.ofMillis(5000));
			
			JedisCluster jedisCluster = new JedisCluster(
				jedisClusterNodes, 
				connectionTimeout, 
				soTimeout,
				maxAttempts,
				password,
				"mypoc",
				jedisPoolConfig,
				true
			);
			
			for (int i=1; i<=150; i++) {
    				String tmpKey = "key" + i;
    				String tmpValue = "Value" + i;
    				jedisCluster.set( tmpKey, tmpValue);
			}
	
			jedisCluster.close();
			
		}catch (Exception e) {
			e.printStackTrace();
		}
		
    }
}

5.2 查看集群状态

使用以下命令获取集群 Key 分布情况：

redis-cli -a 'yourpassword' --tls --cluster call Configuration endpoint:6379 info Keyspace

可以观察到集群的 3 个 Shard 和它们的 Replica 上没有 Key 存在，第一行对应的 Shard 是 my-poc-cluster-redis-0003-001。

5.3 运行写入测试 Code

5.4 再次查看集群状态

使用以下命令获取集群 Key 分布情况：

redis-cli -a 'yourpassword' --tls --cluster call Configuration endpoint:6379 info Keyspace

可以观察到 key 基本平均分布在 3 个 shard 上，至于 Replica 上的 Key，是因为为了保持高可用而复制过去的。同样我们也可以通过 Amazon CloudWatch 观察到，Set Type 的 Command 基本也是平均的。

六、Failover 测试

当 Amazon ElastiCache for Redis 使用集群模式并且为每个 Shard 配置 Replica 时，Jedis 默认是支持 Failover 的。下来让我们简单验证一下。

6.1 测试 Code

创建名为 MyJedisFailover.java 的文件，复制并粘贴以下内容，并运行。

package com.mycompany.app;
import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.Connection;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import java.util.Date;

public class MyJedisFailover
{
	public static void main( String[] args )
    {
		try {
			String host = "Configuration endpoint";
			int port = 6379;
			int connectionTimeout = 10000;
            int soTimeout = 10000;
            int maxAttempts = 20;
			String password = "yourpassword";
			
			Set<HostAndPort> jedisClusterNodes = new HashSet<HostAndPort>();
        	jedisClusterNodes.add(new HostAndPort(host, port));

			GenericObjectPoolConfig<Connection> jedisPoolConfig = new GenericObjectPoolConfig<Connection>();
        	jedisPoolConfig.setMaxTotal(100);
        	jedisPoolConfig.setMaxIdle(10);
        	jedisPoolConfig.setMinIdle(10);
        	jedisPoolConfig.setMaxWait(Duration.ofMillis(5000));
			
			JedisCluster jedisCluster = new JedisCluster(
				jedisClusterNodes, 
				connectionTimeout, 
				soTimeout,
				maxAttempts,
				password,
				"ytpoc",
				jedisPoolConfig,
				true
			);
			
			int i = 1;
			while (true) {
				String tmpKey = "key" + i;
    				String tmpValue = "Value" + i;
    				jedisCluster.set(tmpKey, tmpValue);
				System.out.println("Key: " + tmpKey);
				System.out.println("Value: " + tmpValue);
				Date nowTime = new Date();
        				System.out.println(nowTime);
				i++;
			}
			
		}catch (Exception e) {
			e.printStackTrace();
		}
		
    }
}

6.2 触发 Amazon ElastiCache for Redis Failover

执行下述命令，触发 Shard 1 的 Failover：

aws elasticache test-failover --replication-group-id "my-poc-cluster-redis" --node-group-id "0001"

6.3 测试结果

通过 Amazon CloudWatch 查看 Current Connections 指标，最初每个 node 上都有连接数。

一段时间后，通过 Amazon CloudWatch 查看 Current Connections 指标，发现 my-poc-cluster-redis-0001-001 上没有连接数了，此时已经 Failover 了。

同时，通过 Amazon CloudWatch 查看 Is Master 指标，发现 my-poc-cluster-redis-0001-002 已经变成了 Master。

可以观察到 Failover 期间 Code 没有中断。

七、总结

Jedis 是 Redis 的一款 Java 客户端，专为提高性能和易用性而设计。最新版本支持 Redis 版本 5.0、6.0、6.2 和 7.0。本文带领大家了解了如何使用 Jedis 分别以单机模式，连接池模式和集群模式连接操作 Amazon ElastiCache for Redis，这也同样适用于 Amazon MemoryDB for Redis。同时我们也验证了 Jedis 本身支持的分片功能和 Failover 的能力。关于 Jedis 也有更多的用法，例如 Pipeline 和 Pub/Sub等，您可以参考 Github。