为 StackExchange.Redis 实现最小负载连接的问题

Posted 2023-03-26

【中文标题】为 StackExchange.Redis 实现最小负载连接的问题

【英文标题】：Problem in implementing least loaded connection for StackExchange.Redis

【发布时间】：2021-12-18 01:16:36

【问题描述】：

根据以下答案，我为StackExchange.Redis 实现了least loaded 连接：

https://***.com/a/58106770

public class RedisConnectionWrapper : IRedisConnectionWrapper

    #region Fields

    private readonly Config _config;
    private bool _disposed = false;
    private readonly Lazy<string> _connectionString;
    private static ConcurrentBag<Lazy<ConnectionMultiplexer>> _connections;
    
    #endregion

    #region Ctor

    public RedisConnectionWrapper(Config config)
    
        _config = config;
        _connectionString = new Lazy<string>("CONNECTION_STRING");
        ConnectionMultiplexer.SetFeatureFlag("preventthreadtheft", _config.RedisPreventThreadTheft);

        if (_config.UseLeastLoadedConnection)
            InitializeLeastLoadedConnections();
    

    #endregion

    /// <summary>
    /// Initialize lazy connections to Redis servers
    /// </summary>
    /// <returns></returns>
    private void InitializeLeastLoadedConnections()
    
        _connections = new ConcurrentBag<Lazy<ConnectionMultiplexer>>();

        for (var i = 0; i < _config.PoolSize; i++)
        
            var connection = ConnectionMultiplexer.Connect(_connectionString.Value);

            connection.IncludePerformanceCountersInExceptions = true;

            _connections.Add(new Lazy<ConnectionMultiplexer>(connection));
        
    

    /// <summary>
    /// Get least loaded connection to Redis servers
    /// </summary>
    /// <returns></returns>
    protected ConnectionMultiplexer GetLeastLoadedConnection()
    
        Lazy<ConnectionMultiplexer> connection;

        var loadedLazys = _connections.Where(lazy => lazy.IsValueCreated && lazy.Value.IsConnected);

        if (loadedLazys.Count() == _connections.Count)
        
            var minValue = _connections.Min(lazy => lazy.Value.GetCounters().TotalOutstanding);
            connection = _connections.First(lazy => lazy.Value.GetCounters().TotalOutstanding == minValue);
        
        else
        
            Console.WriteLine("Creating a new connection to Redis");
            connection = _connections.First(lazy => !lazy.IsValueCreated);
        

        return connection.Value;
    

    /// <summary>
    /// Release all resources associated with this object
    /// </summary>
    public void Dispose()
    
        Dispose(true);
        GC.SuppressFinalize(this);
    

    // Protected implementation of Dispose pattern.
    protected virtual void Dispose(bool disposing)
    
        if (_disposed)
            return;

        if (disposing)
        
            if (_config.UseLeastLoadedConnection)
            
                var activeConnections = _connections.Where(lazy => lazy.IsValueCreated).ToList();
                activeConnections.ForEach(connection => connection.Value.Dispose());
            
        
        _disposed = true;
    

但是我在高流量时遇到了这个异常：

System.InvalidOperationException：序列不包含匹配元素 在 System.Linq.ThrowHelper.ThrowNoMatchException() 在 System.Linq.Enumerable.First[TSource](IEnumerable1 源，Func2 谓词)

有人可以帮帮我吗？

【问题讨论】：

这可能是在检查任何未连接的连接然后寻找第一个未连接的连接之间的竞争条件。

当我获得惰性连接时，我正在检查 IsConnected 属性：var loadedLazys = _connections.Where(lazy => lazy.IsValueCreated && lazy.Value.IsConnected); @stuartd

如果您澄清 which First() 调用会引发错误可能会有所帮助

这是第一个发生的原因。我对我的代码做了一些更改，它运行良好。我在下面的答案中描述了变化。感谢您的回复@stuartd

【参考方案1】：

经过一些跟踪，我做了一些更改并且成功了：

    var loadedLazys = _connections.Where(lazy => lazy.IsValueCreated);

    if (loadedLazys.Count() == _connections.Count)
    
       connection = _connections.OrderBy(lazy => lazy.Value.GetCounters().TotalOutstanding).First();
    

我也更改了这部分代码：

    private void InitializeLeastLoadedConnections()
    
        lock (_lock)
        
            _connections = new ConcurrentBag<Lazy<ConnectionMultiplexer>>();
            for (var i = 0; i < _config.PoolSize; i++)
            
                _connections.Add(new Lazy<ConnectionMultiplexer>(() =>
                
                    var connection = ConnectionMultiplexer.Connect(_connectionString.Value);
                    connection.IncludePerformanceCountersInExceptions = true;
                    return connection;
                ));
            
        
    

我还将GetLeastLoadedConnection() 的返回类型从ConnectionMultiplexer 更改为IConnectionMultiplexer。