使用 BookSleeve 的 ConnectionUtils.Connect() 将 SignalR 与 Redis 消息总线故障转移一起使用

Posted 2023-03-04

【中文标题】使用 BookSleeve 的 ConnectionUtils.Connect() 将 SignalR 与 Redis 消息总线故障转移一起使用

【英文标题】：Using SignalR with Redis messagebus failover using BookSleeve's ConnectionUtils.Connect()

【发布时间】：2013-01-31 21:25:25

【问题描述】：

我正在尝试使用 SignalR 应用创建 Redis 消息总线故障转移方案。

起初，我们尝试了一个简单的硬件负载平衡器故障转移，它只监控两台 Redis 服务器。 SignalR 应用程序指向单个 HLB 端点。然后，我在一台服务器上失败了，但在没有回收 SignalR 应用程序池的情况下，无法在第二台 Redis 服务器上成功获取任何消息。这大概是因为它需要向新的 Redis 消息总线发出设置命令。

从 SignalR RC1 开始，Microsoft.AspNet.SignalR.Redis.RedisMessageBus 使用 Booksleeve 的 RedisConnection() 连接到用于发布/订阅的单个 Redis。

我创建了一个新类 RedisMessageBusCluster()，它使用 Booksleeve 的 ConnectionUtils.Connect() 连接到 Redis 服务器集群中的一个。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using BookSleeve;
using Microsoft.AspNet.SignalR.Infrastructure;

namespace Microsoft.AspNet.SignalR.Redis

    /// <summary>
    /// WIP:  Getting scaleout for Redis working
    /// </summary>
    public class RedisMessageBusCluster : ScaleoutMessageBus
    
        private readonly int _db;
        private readonly string[] _keys;
        private RedisConnection _connection;
        private RedisSubscriberConnection _channel;
        private Task _connectTask;

        private readonly TaskQueue _publishQueue = new TaskQueue();

        public RedisMessageBusCluster(string serverList, int db, IEnumerable<string> keys, IDependencyResolver resolver)
            : base(resolver)
        
            _db = db;
            _keys = keys.ToArray();

            // uses a list of connections
            _connection = ConnectionUtils.Connect(serverList);

            //_connection = new RedisConnection(host: server, port: port, password: password);

            _connection.Closed += OnConnectionClosed;
            _connection.Error += OnConnectionError;


            // Start the connection - TODO:  can remove this Open as the connection is already opened, but there's the _connectTask is used later on
            _connectTask = _connection.Open().Then(() =>
            
                // Create a subscription channel in redis
                _channel = _connection.GetOpenSubscriberChannel();

                // Subscribe to the registered connections
                _channel.Subscribe(_keys, OnMessage);

                // Dirty hack but it seems like subscribe returns before the actual
                // subscription is properly setup in some cases
                while (_channel.SubscriptionCount == 0)
                
                    Thread.Sleep(500);
                
            );
        


        protected override Task Send(Message[] messages)
        
            return _connectTask.Then(msgs =>
            
                var taskCompletionSource = new TaskCompletionSource<object>();

                // Group messages by source (connection id)
                var messagesBySource = msgs.GroupBy(m => m.Source);

                SendImpl(messagesBySource.GetEnumerator(), taskCompletionSource);

                return taskCompletionSource.Task;
            ,
            messages);
        

        private void SendImpl(IEnumerator<IGrouping<string, Message>> enumerator, TaskCompletionSource<object> taskCompletionSource)
        
            if (!enumerator.MoveNext())
            
                taskCompletionSource.TrySetResult(null);
            
            else
            
                IGrouping<string, Message> group = enumerator.Current;

                // Get the channel index we're going to use for this message
                int index = Math.Abs(group.Key.GetHashCode()) % _keys.Length;

                string key = _keys[index];

                // Increment the channel number
                _connection.Strings.Increment(_db, key)
                                   .Then((id, k) =>
                                   
                                       var message = new RedisMessage(id, group.ToArray());

                                       return _connection.Publish(k, message.GetBytes());
                                   , key)
                                   .Then((enumer, tcs) => SendImpl(enumer, tcs), enumerator, taskCompletionSource)
                                   .ContinueWithNotComplete(taskCompletionSource);
            
        

        private void OnConnectionClosed(object sender, EventArgs e)
        
            // Should we auto reconnect?
            if (true)
            
                ;
            
        

        private void OnConnectionError(object sender, BookSleeve.ErrorEventArgs e)
        
            // How do we bubble errors?
            if (true)
            
                ;
            
        

        private void OnMessage(string key, byte[] data)
        
            // The key is the stream id (channel)
            var message = RedisMessage.Deserialize(data);

            _publishQueue.Enqueue(() => OnReceived(key, (ulong)message.Id, message.Messages));
        

        protected override void Dispose(bool disposing)
        
            if (disposing)
            
                if (_channel != null)
                
                    _channel.Unsubscribe(_keys);
                    _channel.Close(abort: true);
                

                if (_connection != null)
                
                    _connection.Close(abort: true);
                                
            

            base.Dispose(disposing);
        
    

Booksleeve 有自己的机制来确定主服务器，并且会自动故障转移到另一台服务器，现在正在使用SignalR.Chat 进行测试。

在web.config中，我设置了可用服务器列表：

<add key="redis.serverList" value="dbcache1.local:6379,dbcache2.local:6379"/>

然后在Application_Start():

        // Redis cluster server list
        string redisServerlist = ConfigurationManager.AppSettings["redis.serverList"];

        List<string> eventKeys = new List<string>();
        eventKeys.Add("SignalR.Redis.FailoverTest");
        GlobalHost.DependencyResolver.UseRedisCluster(redisServerlist, eventKeys);

我为Microsoft.AspNet.SignalR.Redis.DependencyResolverExtensions添加了两个额外的方法：

public static IDependencyResolver UseRedisCluster(this IDependencyResolver resolver, string serverList, IEnumerable<string> eventKeys)

    return UseRedisCluster(resolver, serverList, db: 0, eventKeys: eventKeys);


public static IDependencyResolver UseRedisCluster(this IDependencyResolver resolver, string serverList, int db, IEnumerable<string> eventKeys)

    var bus = new Lazy<RedisMessageBusCluster>(() => new RedisMessageBusCluster(serverList, db, eventKeys, resolver));
    resolver.Register(typeof(IMessageBus), () => bus.Value);

    return resolver;

现在的问题是，当我启用了多个断点时，直到添加了用户名，然后禁用所有断点，应用程序才能按预期工作。但是，从一开始就禁用断点，在连接过程中似乎存在一些可能会失败的竞争条件。

因此，在RedisMessageCluster():

    // Start the connection
    _connectTask = _connection.Open().Then(() =>
    
        // Create a subscription channel in redis
        _channel = _connection.GetOpenSubscriberChannel();

        // Subscribe to the registered connections
        _channel.Subscribe(_keys, OnMessage);

        // Dirty hack but it seems like subscribe returns before the actual
        // subscription is properly setup in some cases
        while (_channel.SubscriptionCount == 0)
        
            Thread.Sleep(500);
        
    );

我尝试添加一个Task.Wait，甚至是一个额外的Sleep()（上面未显示）——它们正在等待/等等，但仍然出现错误。

重复出现的错误似乎在Booksleeve.MessageQueue.cs ~ln 71:

A first chance exception of type 'System.InvalidOperationException' occurred in BookSleeve.dll
iisexpress.exe Error: 0 : SignalR exception thrown by Task: System.AggregateException: One or more errors occurred. ---> System.InvalidOperationException: The queue is closed
   at BookSleeve.MessageQueue.Enqueue(RedisMessage item, Boolean highPri) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\MessageQueue.cs:line 71
   at BookSleeve.RedisConnectionBase.EnqueueMessage(RedisMessage message, Boolean queueJump) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\RedisConnectionBase.cs:line 910
   at BookSleeve.RedisConnectionBase.ExecuteInt64(RedisMessage message, Boolean queueJump) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\RedisConnectionBase.cs:line 826
   at BookSleeve.RedisConnection.IncrementImpl(Int32 db, String key, Int64 value, Boolean queueJump) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\IStringCommands.cs:line 277
   at BookSleeve.RedisConnection.BookSleeve.IStringCommands.Increment(Int32 db, String key, Int64 value, Boolean queueJump) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\IStringCommands.cs:line 270
   at Microsoft.AspNet.SignalR.Redis.RedisMessageBusCluster.SendImpl(IEnumerator`1 enumerator, TaskCompletionSource`1 taskCompletionSource) in c:\Projects\Frameworks\SignalR\SignalR.1.0RC1\SignalR\src\Microsoft.AspNet.SignalR.Redis\RedisMessageBusCluster.cs:line 90
   at Microsoft.AspNet.SignalR.Redis.RedisMessageBusCluster.<Send>b__2(Message[] msgs) in c:\Projects\Frameworks\SignalR\SignalR.1.0RC1\SignalR\src\Microsoft.AspNet.SignalR.Redis\RedisMessageBusCluster.cs:line 67
   at Microsoft.AspNet.SignalR.TaskAsyncHelper.GenericDelegates`4.<>c__DisplayClass57.<ThenWithArgs>b__56() in c:\Projects\Frameworks\SignalR\SignalR.1.0RC1\SignalR\src\Microsoft.AspNet.SignalR.Core\TaskAsyncHelper.cs:line 893
   at Microsoft.AspNet.SignalR.TaskAsyncHelper.TaskRunners`2.<>c__DisplayClass42.<RunTask>b__41(Task t) in c:\Projects\Frameworks\SignalR\SignalR.1.0RC1\SignalR\src\Microsoft.AspNet.SignalR.Core\TaskAsyncHelper.cs:line 821
   --- End of inner exception stack trace ---
---> (Inner Exception #0) System.InvalidOperationException: The queue is closed
   at BookSleeve.MessageQueue.Enqueue(RedisMessage item, Boolean highPri) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\MessageQueue.cs:line 71
   at BookSleeve.RedisConnectionBase.EnqueueMessage(RedisMessage message, Boolean queueJump) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\RedisConnectionBase.cs:line 910
   at BookSleeve.RedisConnectionBase.ExecuteInt64(RedisMessage message, Boolean queueJump) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\RedisConnectionBase.cs:line 826
   at BookSleeve.RedisConnection.IncrementImpl(Int32 db, String key, Int64 value, Boolean queueJump) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\IStringCommands.cs:line 277
   at BookSleeve.RedisConnection.BookSleeve.IStringCommands.Increment(Int32 db, String key, Int64 value, Boolean queueJump) in c:\Projects\Frameworks\BookSleeve-1.2.0.5\BookSleeve\IStringCommands.cs:line 270
   at Microsoft.AspNet.SignalR.Redis.RedisMessageBusCluster.SendImpl(IEnumerator`1 enumerator, TaskCompletionSource`1 taskCompletionSource) in c:\Projects\Frameworks\SignalR\SignalR.1.0RC1\SignalR\src\Microsoft.AspNet.SignalR.Redis\RedisMessageBusCluster.cs:line 90
   at Microsoft.AspNet.SignalR.Redis.RedisMessageBusCluster.<Send>b__2(Message[] msgs) in c:\Projects\Frameworks\SignalR\SignalR.1.0RC1\SignalR\src\Microsoft.AspNet.SignalR.Redis\RedisMessageBusCluster.cs:line 67
   at Microsoft.AspNet.SignalR.TaskAsyncHelper.GenericDelegates`4.<>c__DisplayClass57.<ThenWithArgs>b__56() in c:\Projects\Frameworks\SignalR\SignalR.1.0RC1\SignalR\src\Microsoft.AspNet.SignalR.Core\TaskAsyncHelper.cs:line 893
   at Microsoft.AspNet.SignalR.TaskAsyncHelper.TaskRunners`2.<>c__DisplayClass42.<RunTask>b__41(Task t) in c:\Projects\Frameworks\SignalR\SignalR.1.0RC1\SignalR\src\Microsoft.AspNet.SignalR.Core\TaskAsyncHelper.cs:line 821<---



public void Enqueue(RedisMessage item, bool highPri)

    lock (stdPriority)
    
        if (closed)
        
            throw new InvalidOperationException("The queue is closed");
        

正在抛出已关闭队列异常的位置。

我预见到另一个问题：由于 Redis 连接是在 Application_Start() 中建立的，因此在“重新连接”到另一台服务器时可能会出现一些问题。但是，我认为这在使用单数 RedisConnection() 时是有效的，因为只有一个连接可供选择。但是，通过 ConnectionUtils.Connect() 的介绍，我想听听 @dfowler 或其他 SignalR 人员如何在 SignalR 中处理这种情况。

【问题讨论】：

我会看看，但是：发生的第一件事是您不需要调用Open，因为您拥有的连接应该已经打开。不过，我无法立即查看，因为我要准备起飞了

我认为这里有两个问题。 1) Booksleeve 如何处理故障转移； 2) SignalR 如何使用游标来跟踪客户端。初始化新消息总线时，mb1 中的所有游标都不存在于 mb2 上。因此，在重置 SignalR 应用程序池时，它将开始工作 - 而不是之前，这显然不是一个可行的选择。

描述 SignalR 如何使用游标的链接：***.com/questions/13054592/…

尝试使用最新版本的redis消息总线。它支持传入连接工厂，并在服务器宕机时处理重试连接。

你有发行说明的链接吗？谢谢。

【参考方案1】：

SignalR 团队现在已经使用 StackExchange.Redis（BookSleeve 的继承者）实现了对自定义连接工厂的支持，它通过 ConnectionMultiplexer 支持冗余 Redis 连接。

最初遇到的问题是，尽管我在 BookSleeve 中创建了自己的扩展方法来接受服务器集合，但无法进行故障转移。

现在，随着 BookSleeve 向 StackExchange.Redis 的发展，我们现在可以在 Connect 初始化中直接通过 configure 收集服务器/端口。

在创建UseRedisCluster 方法方面，新实现比我走的路要简单得多，并且后端管道现在支持真正的故障转移：

var conn = ConnectionMultiplexer.Connect("redisServer1:6380,redisServer2:6380,redisServer3:6380,allowAdmin=true");

StackExchange.Redis 还允许进行额外的手动配置，如文档的Automatic and Manual Configuration 部分所述：

ConfigurationOptions config = new ConfigurationOptions

    EndPoints =
    
         "redis0", 6379 ,
         "redis1", 6380 
    ,
    CommandMap = CommandMap.Create(new HashSet<string>
     // EXCLUDE a few commands
        "INFO", "CONFIG", "CLUSTER",
        "PING", "ECHO", "CLIENT"
    , available: false),
    KeepAlive = 180,
    DefaultVersion = new Version(2, 8, 8),
    Password = "changeme"
;

本质上，使用一组服务器初始化我们的 SignalR 横向扩展环境的能力现在解决了最初的问题。

【讨论】：

我应该用 500 rep 赏金奖励你的回答吗？ ;)

好吧，如果你相信这就是现在的答案 :)

@ElHaix 既然你问了这个问题，你可能最有资格说你的答案是决定性的还是只是拼图中的一部分——我建议添加一个句子来说明是否以及可能如何它解决了你的问题

所以？赏金？或者我可以等到它引起更多关注。

我是否遗漏了什么，或者这只是在功能分支中，而不是在主 (2.1) nuget 包中？此外，似乎在 bug-stackexchange (github.com/SignalR/SignalR/tree/bug-stackexchange/src/…) 分支中，RedisScaleoutConfiguration 类中还没有办法提供您自己的多路复用器。