从 Redis 连接丢失中恢复答案

【问题标题】：Recovering from Redis connectivity loss从 Redis 连接丢失中恢复
【发布时间】：2019-01-15 21:28:47
【问题描述】：

我正在寻找Redis连接丢失后[在多线程环境中]恢复的参考实现。到目前为止找不到任何有意义的东西。

设置：我有一个 Azure 辅助角色，它在多个线程中运行相同的代码 (ThreadProc)。最初，我有静态 ConnectionMultiplexer 并在每次 Redis 操作之前执行 .GetDatabase() 。那根本没有通过压力测试（一旦负载从低增加到中等，就会出现大量“无法连接”的错误）。我已将其更改为：

static readonly ConnectionMultiplexer _connection = ConnectionMultiplexer.Connect(...);
static readonly IDatabase _cache = _connection.GetDatabase();

void ThreadProc() // running in multiple threads
{
    while (true)
    {
      // using _cache here
    }
}

即使在高负载下（每个工作角色实例 1000+ 操作/秒）也能正常工作，直到我得到“没有可用的连接来服务此操作”，此后事情就无法恢复。

请让我知道可以从间歇性连接问题中恢复的正确/推荐代码是什么。

【问题讨论】：

我遇到了完全相同的问题（就在我去度假之后，natch）。我的“解决方案”是显而易见且乏味的，用 try/catch 包装对 redis 的每次调用，在每次失败后重新创建连接，然后重试初始操作。还没有从假期回来足够长的时间来看看它是如何工作的:-)。
Ken，你是否将每个 _cache.* 调用都包装在 lock() 中？或者如何在多线程环境中安全地重新创建共享连接？
在下面查看我的答案。让我知道您是否担心它没有解决的情况。（很高兴改进我的代码！）

标签： c# redis stackexchange.redis

【解决方案1】：

编辑 2015-05-02： 虽然 StackExchange.Redis 客户端的更高版本明确应该在内部和自动处理这种“丢失的连接”->“重新连接”逻辑，但我的测试已经毫无疑问地表明他们无法成功地完成它，因此仍然需要这种事情，至少在繁忙的环境中是这样。我将下面的代码从我的缓存层中抽出一段时间，结果出现了数以万计的连接失败错误。我把它放回去，这些都消失了。

编辑 2015-02-24： 不再需要此方法。 StackExchange.Redis 客户端的最新版本正确处理断开连接 - 它们会自动重新连接，下面的解决方法只会干扰事情。出于历史目的将其保留在这里，但我的建议是忽略它。

以下是我的 SimpleCacheRedis<T> 包装器中的一些方法，它们展示了我如何处理问题：

public async Task<TValue> GetAsync(string key, Func<Task<TValue>> missingFunc)
{
    key = GetKey(key);
    var value = default(TValue);
    try
    {
        var db = _connection.GetDatabase();
        var str = await db.StringGetAsync(key);
        if (!str.IsNullOrEmpty)
        {
            value = _jsonSerializer.Deserialize<TValue>(str);
        }
    }
    catch (RedisConnectionException ex)
    {
        HandleRedisConnectionError(ex);
    }
    catch (Exception ex)
    {
        _logger.Error("Error retrieving item '" + key +
                      "' from Redis cache; falling back to missingFunc(). Error = " + ex);
    }
    if (value == default(TValue))
    {
        present = false;
        value = await missingFunc();
        await PerformAddAsync(key, value);
    }
    return value;
}

private void HandleRedisConnectionError(RedisConnectionException ex)
{
    _logger.Error("Connection error with Redis cache; recreating connection for the next try, and falling back to missingFunc() for this one. Error = " + ex.Message);
    Task.Run(async () =>
    {
        try
        {
            await CreateConnectionAsync();
        }
        catch (Exception genEx)
        {
            _logger.Error("Unable to recreate redis connection (sigh); bailing for now: " + genEx.Message);
        }
    });
}

private async Task CreateConnectionAsync()
{
    if (_attemptingToConnect) return;
    var sw = new StringWriter();
    try
    {
        _attemptingToConnect = true;
        _connection = await ConnectionMultiplexer.ConnectAsync(_redisCs, sw);
    }
    catch (Exception ex)
    {
        _logger.Error("Unable to connect to redis async: " + ex);
        _logger.Debug("internal log: \r\n" + sw);
        throw;
    }
    finally
    {
        _attemptingToConnect = false;
    }
}

基本上，如果我发现由于 RedisConnectionException 而无法连接到 Redis，我会分拆一个单独的 async 任务来重新创建共享连接。当然，那个调用可能会失败，但无论如何，在那段时间调用都会失败。一旦成功，任何新调用都将使用该新（重新）创建的连接。就像我上面说的，有点无聊。

我的情况可能与你的情况有些不同，因为我没有将 Redis 用作永久存储，而只是用作缓存。这意味着丢失 redis 连接的唯一影响是我需要从数据库而不是缓存中检索结果。所以我可以稍微宽松地对待某些事情。

【讨论】：

不错的解决方案！感谢分享。
这还是SS无法处理重连的情况吗？
我不想批评您的代码，但它并不能干净地处理并行性，对吗？由于不涉及锁定，因此两个并行的 get 调用可能会导致创建两个连接，其中一个将不再被引用。虽然我不确定这是否真的是个问题；）
@Pharao2k - 这是一个有效的观点：你绝对可以让两个线程同时检查 _attemptingToConnect 变量，看看它没有设置，然后都尝试建立连接，其中一个会得到孤立的，从未使用过且从未关闭过（至少根据我的代码）。某种锁可以解决这个问题。另一方面，除非你有一个非常繁忙的服务器，否则它不会经常发生，我认为孤立连接的后果很小，而且我喜欢更简单的代码，不会受到导航锁的复杂性的影响。仍然可能值得做......

【解决方案2】：

好吧，如果没有人愿意，我想我会回答我自己的问题，虽然这看起来很奇怪，作为一个基本的用例。

这是管理连接丢失的类：

static class RedisConnectionManager
{
    private static readonly Dictionary<string, IDatabase> _dictionary = new Dictionary<string, IDatabase>();

    internal static IDatabase GetDatabase(string connectionString)
    {
        lock (_dictionary)
        {
            if (!_dictionary.ContainsKey(connectionString))
                _dictionary.Add(connectionString, ConnectionMultiplexer.Connect(connectionString).GetDatabase());
            if (!_dictionary[connectionString].Multiplexer.IsConnected)
            {
                _dictionary[connectionString].Multiplexer.Dispose();
                _dictionary[connectionString] = ConnectionMultiplexer.Connect(connectionString).GetDatabase();
            }
            return _dictionary[connectionString];
        }
    }
}

这个类处理多个连接字符串，所以如果你只有一个，代码会更简单。请注意明确的Multiplexer.Dispose() 呼叫。由于底层对象拥有物理 TCP 连接，因此您不能等到 GC 启动才释放资源。到那时，根据您的负载，您可能会有数千个孤立的 TCP 连接。

此代码运行良好，但我仍然不能 100% 确定这是处理此问题的最佳方法。如果有人知道如何改进这一点，请告诉我。

【讨论】：