条条大路通罗马 – 如何在.NET程序中使用StackExchange.Redis操作Amazon ElastiCache for Redis

一． 前言

1.1 关于Amazon ElastiCache for Redis

Amazon ElastiCache 是一种 Web 服务，可让用户在云中轻松设置、管理和扩展分布式内存数据存储或缓存环境。它可以提供高性能、可扩展且具有成本效益的缓存解决方案。Amazon ElastiCache 使用 Redis 和 Memcached 引擎。

ElastiCache for Redis 使用 Redis，是在开源 Redis 上构建，并与 Redis API 兼容，既可与您的 Redis 客户端协作，也可使用开放的 Redis 数据格式存储数据。无需更改任何代码，您自行管理的 Redis 应用程序可与 ElastiCache for Redis 无缝协作。

ElastiCache for Redis 具有多种功能，他的主要特点如下：

• 自动检测缓存节点故障并从中恢复。
• 在支持复制的 Redis 集群中，主集群故障转移到只读副本的多可用区。
• Redis（已启用集群模式）支持跨多达 500 个分区对数据进行分区。
• 对于 Redis 3.2 及更高版本，所有版本都支持传输中加密和使用身份验证的静态加密。此支持可帮助您构建符合 HIPAA 要求的应用程序。
• 在可用区中灵活放置节点和集群，以提高容错能力。
• 与其他 AWS 服务集成，例如 Amazon EC2、Amazon CloudWatch、AWS CloudTrail 和 Amazon SNS。此集成有助于提供高性能和高安全性的托管式内存中的缓存解决方案。
• ElastiCache for Redis 可以管理备份、软件修补、自动故障检测和恢复。
• 您可以在需要时执行自动备份，也可以手动创建您自己的备份快照。您可以使用这些备份还原集群。ElastiCache for Redis 还原过程可靠且高效。

1.2 Amazon ElastiCache for Redis 部署模式

Redis 部署模式主要可以分为集群模式和⾮集群模式，⾮集群模式⼜可以分为单节点模式和单分⽚多节点模式（最多拥有6个节点），启⽤集群模式后可以最多拥有500个分⽚。

⼀个分⽚（节点组）中⼀个节点充当主节点，可以进⾏读写，其他节点充当主节点的只读副本。

1.3 关于StackExchange.Redis

StackExchange.Redis 是⽤于 .NET 语⾔（C# 等）的⾼性能通⽤ Redis 客户端。

功能特点：

• 高性能多路复用设计，允许高效使用来自多个调用线程的共享连接；
• Redis 节点配置的抽象：客户端可以静默协商多个 Redis 服务器以实现健壮性和可用性；
• 方便地访问完整的 redis 功能集；
• 同步和异步使用的完全双重编程模型，不需要 TPL；
• 支持redis“集群”模式；

有多种方式可以安装 StackExchange.Redis，包括：

• 使用.NET CLI:

dotnet add package StackExchange.Redis

• 使用包管理器控制台：

PM> Install-Package StackExchange.Redis

• 在Visual Studio中使用NuGet GUI

⼆． 部署和应⽤测试

2.1 准备⼯作

在开始准备部署和开发测试之前，我们需要准备如下资源：

1. 带有公有子网和私有子网的VPC资源(ElastiCache 不能被公网访问)；
2. Windows实例1台部署在公有子网，并安装了Visual Studio进行.NET应用程序开发测试工作（本文使用VS2022，ASP.NET Web API，.NET 6.0；
3. 创建Asp.Net项目“ElastiCacheClient”，并使用NuGet添加“StackExchange.Redis”到项目中；

创建一个模拟测试的Controller，包含了2个动作：添加待办事项（CreateTodoItem）和查询待办事项（GetTodoItem）。

public class TodoItemsController
{  
  private ConnectionMultiplexer redis;
  private IDatabase db ;

  # 构造函数
  public TodoItemsController()
  {
  }
}

2.2连接到ElastiCache

端点（Endpoint）是我们连接ElastiCache的唯一方式，我们前面介绍了ElastiCache的部署模式分为集群和非集群，连接到ElastiCache的端点我们分成了三种情况：

1. Redis 独立节点：使用该节点的终端节点进行读取和写入操作。

2. Redis（已禁用集群模式）集群，使用主端点执行所有写入操作。使用读取器终端节点 将在所有只读副本之间均匀地分配指向终端节点的传入连接。使用单独的节点终端节点 进行读取操作（在 API/CLI 中，它们被称作读取终端节点）。

3. Redis（已启用集群模式）集群，使用集群的配置端点执行所有支持已启用集群模式命令的操作。您必须使用支持 Redis 集群的客户端 (Redis 3.2)。您仍可以从独立的节点终端节点进行读取（在 API/CLI 中，它们被称作读取终端节点）。

如果想了解如何获取端点信息，请参考：查找连接端点。

在StackExchange.Redis的控制器的代码中，首先在需要初始化Redis连接器，连接器的初始化主要由核心类ConnectionMultiplexer完成，需要在其Connect方法中指定Endpoint。您可以在在Connect方法中指定多个Endpoint，需要使用逗号分隔，StackExchange.Redis 将自动根据需要判断读写的节点。如果您在读取Redis中的数据的时候想优先从已经配置的只读节点中读取数据，可以使用‘CommandFlags.PreferReplica’参数来指定优先从只读副本读取；但如果只读副本不存在或者不可用，将会从主节点继续读取数据。

接下下，我们对这上面介绍的3种端点的连接方式进行详细的介绍。

2.2.1连接到Redis 独立节点

独立节点意味着我们只有一个分片一个节点，承担了所有读写操作。这个时候，我们在ConnectionMultiplexer的Connect方法中传入端点地址即可。

可以在控制台中查看集群的主节点和只读节点的端点地址（只有一个节点，读写均为该节点，但是有2个地址）；另外还可以在节点列表中找到对应节点的端点地址。

  public TodoItemsController()
  {
			
	// 使用单个node的情况下进行连接
      //redis = ConnectionMultiplexer.Connect("redis-node-001.cnn1.cache.amazonaws.com.cn:6379"); 
}

2.2.2 连接到Redis（已禁用集群模式）集群

已禁用集群模式的集群端点，是一种只有一个分片（Shard）的集群，它具有多个节点（Node）；其中有一个节点是主节点，可以承担读写的任务，而其他节点为副本节点（只承担读任务）。这个时候我们可以在ConnectionMultiplexer的Connect方法中传入多个Node的Endpoint信息，连接器将自动分辨主节点。

端点查找的方法和单节点中的类似，在集群信息中或者节点信息中均可以查找到对应的主节点端点和只读节点的端点信息。

  public TodoItemsController()
  {
   // 多个节点情况下，使用逗号分隔不同节点的地址，连接器将自动区分主节点
      redis = ConnectionMultiplexer.Connect("redis.xxx.cache.amazonaws.com.cn:6379,redis-ro.xxx.cache.amazonaws.com.cn:6379");  
}

2.2.3 连接到Redis（已启用集群模式）集群

针对启用了集群模式的ElastiCache集群，具有多个分片（Shard），每个分片又通常具有多个节点（Node）。我们可以通过配置端点连接到ElastiCache for Redis。

  public TodoItemsController()
  {
	// 开启集群模式，使用Configuration endpoint
      //redis = ConnectionMultiplexer.Connect("redis-cluster.clustercfg.cnn1.cache.amazonaws.com.cn:6379");
      db = redis.GetDatabase();
  }

2.2.4 完整的示例代码

using Microsoft.AspNetCore.Mvc;
using StackExchange.Redis;

namespace ElastiCacheClient
{
    [Route("api/[controller]")]
    [ApiController]
    public class TodoItemsController : ControllerBase
    {

        private ConnectionMultiplexer redis;
        private IDatabase db ;

        public TodoItemsController()
        {
            redis = ConnectionMultiplexer.Connect("xxx.cnn1.cache.amazonaws.com.cn:6379,xxx-ro.cnn1.cache.amazonaws.com.cn:6379");
            db = redis.GetDatabase();
        }

        // GET: api/TodoItems/5
        [HttpGet("{id}")]
        public async Task<ActionResult<String>> GetTodoItem(long id)
        {
            String todo = db.StringGet(id.ToString(), CommandFlags.PreferReplica);

            return todo;
        }

        // POST: api/TodoItems
        [HttpPost]
        public async Task<ActionResult<TodoItemDTO>> CreateTodoItem(TodoItemDTO todoItemDTO)
        {

            db.StringSet(todoItemDTO.Id.ToString(), todoItemDTO.Name.ToString());

            return todoItemDTO;
        }

    }
}

2.3 使⽤⾮加密⽅式连接ElastiCache⾮集群模式

为了帮助确保数据安全，Amazon ElastiCache 和 Amazon EC2 提供了禁止未经授权来访问服务器上数据的机制。通过传输中加密功能，ElastiCache 为您提供了在不同位置之间移动数据时用来保护数据的工具。例如，您可能从复制组中的主节点向只读副本节点移动数据，或在复制组与应用程序之间移动数据。

传输中加密是可选功能，只能在创建 Redis 复制组时在复制组中启用。在创建复制组时，可通过将参数 TransitEncryptionEnabled设置为 true（CLI：–transit-encryption-enabled）在复制组中启用传输中加密。

在使用控制台创建ElastiCache Redis的过程，我们可以勾选“Encryption in transit”来开启加密方式访问SSL。

2.3.1 ElastiCache 传输中加密可实现以下功能：

• 加密连接 – 服务器和客户端连接均采用安全套接字层 (SSL) 加密。
• 加密复制 – 对在主节点与副本节点之间移动的数据进行加密。
• 服务器身份验证 – 客户端可通过身份验证确定它们连接到正确的服务器。
• 客户端身份验证 – 使用 Redis AUTH 功能，服务器可以对客户端进行身份验证。

2.3.2传输中加密的条件：

在规划实现时，应记住有关 Amazon ElastiCache 传输中加密的以下限制：

• 在运行以下 Redis 版本的复制组上支持传输中加密：3.2.6、4.0.10 和更高版本。
• 只有在 Amazon VPC 中运行的复制组支持传输中加密。
• 只有运行以下节点类型的复制组才支持传输中加密。
  • R6g、R5、R4、R3
  • M6g、M5、M4、M3
  • T4g、T3、T2

更多关于加密传输的所需条件请查看：传输中加密 (TLS)

2.3.3 如何在StackExchange.Redis中使⽤传输中加密（SSL）：

如果需要使⽤SSL加密，只需要指定“ssl=true”即可；如果指定了访问密码，则需要通过“password=xxxxxx”来指定访问密码。

public TodoItemsController()
{
    redis = ConnectionMultiplexer.Connect("master.redis.cnn1.cache.amazonaws.com.cn:6379,ssl=true,password=zk-stackexchange-redis-user-access-pwd");
    db = redis.GetDatabase();
}

2.4 常见Key/Value使用

// 设置Key和对应的String值
db.StringSet("mykey", "myvalue");

// 删除Key和对应的值
db.KeyDelete("abc");

// 生成随机Key
string someKey = db.KeyRandom();

StackExchange.Redis 通过 RedisKey 类型表示键。 好消息是，可以从 string 和 byte[] 的隐式转换，允许使用文本和二进制密钥，没有任何复杂性。

例如，如果你只知道Key值想对Value进行更新，可以使用StringIncrement 方法。

db.StringSet("mykey", "123");
db.StringGet("mykey");// 将输出123
db.StringIncrement("mykey");
db.StringGet("mykey");// 将输出124

StackExchange也支持集群模式，会根据get或者set操作的key值自动进行判断使用哪个分片。但如果想集群模式下手工指定分片，需要在get 和 set的key中指定{tag}， tag值将被hash后判断分布在哪个分片上：

// 集群中Key&Value操作
String inputKey = Console.ReadLine();
db.StringSet("{001}"+inputKey, "111");
db.StringSet("{002}"+inputKey, "112");
db.StringSet("{003}"+inputKey, "113");
db.StringIncrement(inputKey);
String result1 = db.StringGet("{001}" + inputKey);
Console.WriteLine(result1);
String result2 = db.StringGet("{002}" + inputKey);
Console.WriteLine(result2);
String result3 = db.StringGet("{003}" + inputKey);
Console.WriteLine(result3);
Console.ReadLine();

2.5 只读模式

ElastiCache for Redis的读写都是在主节点进行，如果只需要只读可以放在副本节点进行。副本节点可以在创建时在控制台或者CLI命令中启用只读副本，也可以在创建完集群后在控制台添加副本。

在StackExchange.Redis启用从只读副本中读取需要做的步骤分2步：

1. 在连接中添加只读副本的endpoint地址；
2. 在读取代码中，使用CommandFlags.PreferReplica优先从只读副本读取数据；

// 根据key获取 string value 并指定优先从读副本读取 
String todo = db.StringGet(id.ToString(), CommandFlags.PreferReplica);

2.6 Pipelining

使用Pipleline 我们可以异步发送消息，而无需等待Redis消息完成才进行下一步的处理。

var aPending = db.StringGetAsync("a");
var bPending = db.StringGetAsync("b");
var a = db.Wait(aPending);
var b = db.Wait(bPending);

需要注意的是，我们这里使用db.Wait，它将自动应用配置的同步超时，但如果您愿意，也可以使用aPending.Wait()或Task.WaitAll(aPending,bPending)。

另外，Pipelining也支持集群模式，但是需要和正常的get和set的集群操作一样，集群下会根据key值判断分布在哪个集群分片上。同时，如果想人工指定相同分片，我们也可以在key中指定分区（片）键，需要在get 和 set的key中指定{tag}， tag值将被hash后计算分布在哪个分片上。

例如：

private static void pipeliningInCluster()
{
 //批量设置参数和值
var setPending = db.StringSetAsync("{keys}pipe555", "this is pipe 555");
db.Wait(setPending);

//读取Key值信息
var getPending = db.StringGetAsync("{keys}pipe555");
RedisValue redisValue = db.Wait(getPending);
Console.WriteLine(redisValue);
Console.ReadLine();
}

2.7  Fire and Forget

当我们不关⼼具体的操作什么时候返回时，我也可以使用‘Fire and Forget’功能，他会立即给我们⼀个返回值，尽管这个返回值是无意义的（因为是异步操作，并不代表完成操作），然后进⼊后台队列进行操作。

db.KeyExpire(todoItemDTO.Id.ToString(), TimeSpan.FromMinutes(5), flags: CommandFlags.FireAndForget);
var value = (string)db.StringGet(todoItemDTO.Id.ToString());

2.8 发布/订阅消息顺序

Redis 发布订阅 (pub/sub) 是一种消息通信模式：发送者 (pub) 发送消息，订阅者 (sub) 接收消息。Redis 客户端可以订阅任意数量的频道。

下图展示了频道 channel1 ， 以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系：

当有新消息通过 PUBLISH 命令发送给频道 channel1 时， 这个消息就会被发送给订阅它的三个客户端：

在StackExchange.Redis中使用 pub/sub API 时，需要决定是否应按顺序处理来自同一连接的消息或并发同时处理消息。按顺序处理它们意味着您不需要过于担心线程安全，并且意味着您保留事件的顺序，但也因此意味着消息会有更多的延迟。

var channel = multiplexer.GetSubscriber().Subscribe("messages");
channel.OnMessage(message =>
{
    Console.WriteLine((string)message.Message);
});

另一种选择是并发处理。 这将不保证处理工作的先后顺序，但它可以显着更快且更具可扩展性。

multiplexer.GetSubscriber().Subscribe("messages", (channel, message) => {
    Console.WriteLine((string)message);
});

Redis 集群支持Publish/Subscribe，但发布/订阅消息是在整个集群中广播，而与订阅特定频道/模式无关。发布行为会连接到集群中的所有节点，而不用客户端连接到所有节点都会收到订阅消息。

2.9 mget / mset 支持

使用StackExchange.Redis 可以一次性读取多个Key的值，这将节约很多的网络消耗提升效率；

批量读取数据可以使用StringGet 或使用 StringGetAsync的重载方法传入RedisKey[]进行读取；

批量写入可以使用StringSet()的重载方法，传入 KeyValuePair() 的数组类型，进行批量的数据设置；

// 初始化key value 数组，用于批量设置
KeyValuePair<RedisKey, RedisValue>[] keyValuePairs = new KeyValuePair<RedisKey, RedisValue>[3];
keyValuePairs[0] = new KeyValuePair<RedisKey, RedisValue>( new RedisKey("555"), new RedisValue( "this is 555"));
keyValuePairs[1] = new KeyValuePair<RedisKey, RedisValue>(new RedisKey("666"), new RedisValue("this is 666"));
keyValuePairs[2] = new KeyValuePair<RedisKey, RedisValue>(new RedisKey("777"), new RedisValue("this is 777"));
//批量设置参数和值
db.StringSet(keyValuePairs);

//设置多个需要查询的Key值信息
RedisKey[] redisKeys = new RedisKey[] { "555", "666", "777" };
//一次读取多个Key值信息
RedisValue[] redisValue = db.StringGet(redisKeys);
// 遍历输出获取到的多个值
foreach (var item in redisValue)
{
Console.WriteLine(item);
}
Console.ReadLine();

在集群模式下，使用mset、mget不支持跨分片操作，因此需要在key中添加{tag}, 这样redis将使用{}中的tag计算hash后判断将数据存储在哪个分片上。

//在集群中批量设置和读取数据
// 初始化key value 数组，用于批量设置
KeyValuePair<RedisKey, RedisValue>[] keyValuePairs = new KeyValuePair<RedisKey, RedisValue>[3];
keyValuePairs[0] = new KeyValuePair<RedisKey, RedisValue>(new RedisKey("{keys}555"), new RedisValue("this is 555k"));
keyValuePairs[1] = new KeyValuePair<RedisKey, RedisValue>(new RedisKey("{keys}666"), new RedisValue("this is 666k"));
keyValuePairs[2] = new KeyValuePair<RedisKey, RedisValue>(new RedisKey("{keys}777"), new RedisValue("this is 777k"));
//批量设置参数和值
db.StringSet(keyValuePairs);

//设置多个需要查询的Key值信息
RedisKey[] redisKeys = new RedisKey[] { "{ keys }555", "{ keys }666", "{ keys }777" };
//一次读取多个Key值信息
RedisValue[] redisValue = db.StringGet(redisKeys);
// 遍历输出获取到的多个值
foreach (var item in redisValue)
{
Console.WriteLine(item);
}

2.10 事务

可以在StackExchange.Redis使用Redis事务，例如：

var newId = CreateNewId();
var tran = db.CreateTransaction();
tran.AddCondition(Condition.HashNotExists(custKey, "UniqueID"));
tran.HashSetAsync(custKey, "UniqueID", newId);
bool committed = tran.Execute();

如果在集群模式下使用事务，只支持在一个事务中进行相同分片（shard）中的Key Value操作，例如如下代码，如果同时设置2个key，必须使用slot tag 将两个key限制在相同的slot中才可以提交：

var custKey = "{001}clusterkey101";
var tran = db.CreateTransaction();
tran.AddCondition(Condition.KeyNotExists(custKey));
tran.StringSetAsync("{001}clusterkey101", "001");
tran.StringSetAsync("{001}clusterkey102", "002");
bool committed = tran.Execute();
Console.WriteLine(committed);
Console.ReadLine();

三． 集群环境故障转移

3.1 环境准备

3.1.1 ElastiCache环境准备：

• Multi-AZ：enabled
• Engine version：6.2.6
• Number of nodes：9
• Shards：3
• Auto-failover：Enabled
• Node type：cache.r6g.large
• Cluster mode：On

3.1.2 .NET framework代码准备：

创建.NET framework c# application console 应用程序用于集群访问的测试，使用Nuget添加‘StackExchange.Redis’最新版本的引用。

在Program类中的Main方法中添加如下代码，该代码将持续30分钟对Redis集群进行不间断的请求，请求间隔为300毫秒，如果发生异常会输出异常信息并继续循环。同时为了保障集群不同分片均收到访问，我们给StringGet设置了不同的{tag}值，让每次的请求可以分布到不同的分片。

private static void loopToTestRedisClusterFallover(int id) {

    db.StringSet(id.ToString(), 0);

    DateTime currentDT = DateTime.Now;
    DateTime targetDT = currentDT.AddMinutes(30);
    while (new DateTime() <= targetDT)
    {
        try
        {
            Console.WriteLine(db.StringGet("{001}"+ id.ToString()).ToString(), CommandFlags.PreferReplica);
            Console.WriteLine(db.StringGet("{002}"+ id.ToString()).ToString(), CommandFlags.PreferReplica);
            Console.WriteLine(db.StringGet("{003}"+ id.ToString()).ToString(), CommandFlags.PreferReplica);

            db.StringIncrement("{001}" + id.ToString());
            db.StringIncrement("{002}" + id.ToString());
            db.StringIncrement("{003}" + id.ToString());

            //每间隔300毫秒进行休息
            Thread.Sleep(300);
            Console.WriteLine("Success time:" + DateTime.Now.ToString());
        }
        catch (Exception ex)
        {
            Console.WriteLine("Failed time:" + DateTime.Now.ToString());
            Console.WriteLine(ex.ToString());
        }
    }

    Console.WriteLine(db.StringGet("{001}" + id.ToString()).ToString(), CommandFlags.PreferReplica);
    Console.WriteLine(db.StringGet("{002}" + id.ToString()).ToString(), CommandFlags.PreferReplica);
    Console.WriteLine(db.StringGet("{003}" + id.ToString()).ToString(), CommandFlags.PreferReplica);

3.1.3 Redis监控实例准备

在ElastiCache for Redis同VPC中启用EC2实例，用于监控Redis的负载情况。（需要注意的是，如果启用SSL加密连接，需要安装和配置Stunnel才可以在CLI中连接到Redis。

# 1. 安装并编译redis-cli
# 2. 安装并配置stunnel（如未开启SSL可跳过此步骤）
第1&2步骤可以参见官方介绍：传输中加密 (TLS)

3.2 测试故障转移

接下来我们可以开始进集群故障转移的测试，整个故障转移的测试步骤我们可以分为三个主要阶段：

a.     启动集群监控;

b.     集群故障转移操作；

c.     故障转移后恢复情况信息收集和对比；

3.2.1 启动命令行对集群进行监控

启动对集群中需要进行failover的分片（shard 0001 ）中3个node（001、002、003）的监控，如果分片被执行故障转移后，监测其中的3个节点的状态变化。

针对节点监控的参考命令如下：

# 1. SSH 连接到EC2实例（建议Linux2）
ssh -i "xxx.pem" ec2-user@xxx.cn-north-1.compute.amazonaws.com.cn
# 切换至3.1.3编译的redis-cli程序
cd redis-stable/
# 执行redis-cli命令连接至Redis节点(如开启SSL并设置了密码）
src/redis-cli -c -h [ENDPOINT] -p 6379 -a '[PASSWORD]'
# 执行redis-cli命令连接至Redis节点(如未开启SSL）
src/redis-cli -c -h [ENDPOINT] -p 6379
# 开启监控
Monitor

3.2.2 启动 .NET 控制台命令程序

通过应用程序的不断请求集群分片中的数据，监控程序的异常变化信息。设置集群连接器的Endpoint，并执行3.1.2代码中的loopToTestRedisClusterFallover测试方法：

启动程序后，可以看到控制台应用程序会每次成功会输出3个数据并带有时间信息:

3.2.3 执行故障转移

进⼊控制台程序，进入集群管理页面，选择将主节点 ‘zk-stackexchange-redis-ssl-003’ 3.1中被监控的Shard进⾏开启故障转移：

3.2.4 观测/对比故障信息

我们可以从3个角度分别观测故障转移和恢复的具体情况：

a.  应用程序的故障和恢复情况；

b.  集群节点监控的故障和恢复的情况；

c. ElastiCache的Event日志观察故障和恢复的情况；

一、应用程序故障和恢复情况

通过观察应用程序Console输出的信息，可以观测到监控服务连接关闭：控制台应用程序在访问redis 集群过程中17:13:57 为故障转移过程中首次失败的时间，17:14:19为故障转移后恢复正常使用的时间。具体日志摘要如下：

二、通过集群节点监控的故障和恢复的情况

通过对集群分片0001的监控我们发现，002节点在故障转移时失败时候的时间戳为：1662714837，对应的GMT+8的时间为：2022-9-9 17:13:57。

003节点为新的主节点，而正式成为主节点的时间戳为：1662714857，对应的GMT+8的时间为：2022-9-9 17:14:17。

三、通过ElastiCache的Event日志观察故障和恢复的情况；

最后，我们也可以通过 “describe-events” 命令查看集群变化情况，需要注意的是，因为时间默认是GTM时间，如转换成当前中国时区需要+8小时；

• 集群在2022-09-09 17:13:56接收到连接中断的请求；
• 集群在2022-09-09 17:14:35完成切换到003节点；
• 集群切换完主从后，原节点并未立即可用，2022-09-09 09:21:57完成了原002节点的恢复使用；

3.3 测试结论

通过整个测试过程对故障转移进行监控的三个场景的日志中，我们可以发现：

• 在主从故障切换的过程中，需要大概3022秒左右的时间来完成切换过程；即可实现集群故障转移，恢复正常使用；
• 集群事件日志中可以看到整个故障转移过程大约39秒左右完成切换后，，但实际集群在22多秒即可恢复使用；
• 原故障副本并非立即可用，需要等待完全恢复后可用；节点会在故障转移完成后开始尝试恢复节点使用；

四． 总结

本文带领大家在.NET Framework中使用C#语言结合‘StackExchange.Redis’框架进行了ElastiCache for Redis的基本使用操作和集群模式的操作演示以及故障切换演练的测试。通过实验我们可以发现在.NET框架下使用StackExchange.Redis框架对ElastiCache for Redis的非集群和集群的支持性是比较友好，使用起来和自建的Redis集群和非集群模式比较一致。

限于篇幅有限，我们只针对常用的使用场景进行了介绍，如果您想了解更多的关于‘StackExchange.Redis‘的介绍，可以参考官方文档以及源码介绍。另外如果您想使用更多的便捷开发功能（比如习惯使用配置文件进行初始化Redis连接）您可以了解下“StackExchange.Redis.Extensions”。

最后，希望本文对您有帮助。

相关博客

• 条条大路通罗马 — 使用 redisson 连接 Amazon ElastiCache for redis 集群：https://aws.amazon.com/cn/blogs/china/connecting-amazon-elasticache-for-redis-cluster-using-redisson/
• 条条大路通罗马 — 使用 redis-py 访问 Amazon ElastiCache for redis 集群：https://aws.amazon.com/cn/blogs/china/use-redis-py-to-access-amazon-elasticache-for-redis-cluster/
• 条条大路通罗马 — 使用 go-redis 连接 Amazon ElastiCache for Redis 集群：https://aws.amazon.com/cn/blogs/china/all-roads-lead-to-rome-use-go-redis-to-connect-amazon-elasticache-for-redis-cluster/
• 条条大路通罗马 — 使用 Hiredis-cluster 连接 Amazon ElastiCache for Redis 集群：https://aws.amazon.com/cn/blogs/china/all-roads-to-rome-series-connect-amazon-elasticache-for-redis-cluster-with-hiredis-cluster/
• 条条大路通罗马 — 使用 Jedis 访问 Amazon ElastiCache for Redis 集群：https://aws.amazon.com/cn/blogs/china/accessing-an-amazon-elasticache-for-redis-cluster-using-jedis/

本篇作者