SignalR 앱을 사용하여 Redis 메시지 버스 장애 조치 시나리오를 만들려고합니다.
처음에는 두 개의 Redis 서버를 모니터링하는 간단한 하드웨어로드 밸런서 장애 조치를 시도했습니다. SignalR 애플리케이션은 단일 HLB 엔드 포인트를 가리 켰습니다. 그런 다음 하나의 서버에 오류가 발생했지만 SignalR 앱 풀을 재활용하지 않고 두 번째 Redis 서버에서 메시지를 성공적으로 가져올 수 없습니다. 아마도 이것은 새로운 Redis 메시지 버스에 설정 명령을 발행해야하기 때문일 것입니다.
SignalR RC1부터는 Booksleeve를 Microsoft.AspNet.SignalR.Redis.RedisMessageBus
사용하여 게시 / 구독 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()
선택할 수있는 연결이 하나 뿐인 singular를 사용할 때 이것이 유효하다고 생각합니다 . 그러나이 시나리오가 SignalR에서 어떻게 처리되는지에 대해 다른 SignalR 직원들 ConnectionUtils.Connect()
로부터 듣고 싶습니다 @dfowler
.
답변
SignalR 팀은 이제 ConnectionMultiplexer를 통한 중복 Redis 연결을 지원하는 BookSleeve의 후속 제품인 StackExchange.Redis 를 사용하여 사용자 지정 연결 팩토리에 대한 지원을 구현했습니다 .
처음 발생한 문제는 BookSleeve에서 서버 모음을 수락하기 위해 자체 확장 메서드를 만들었지 만 장애 조치가 불가능하다는 것입니다.
이제 BookSleeve가 StackExchange.Redis로 발전함에 따라 이제 초기화 에서 바로 서버 / 포트 컬렉션을 구성 할 수 있습니다 Connect
.
새로운 구현은 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 확장 환경을 초기화하는 기능은 이제 초기 문제를 해결합니다.