SignalR核心 - 使死连接无效

Question

问题

我正在使用带有ASP.NET Core SignalR的.NET Core 2.2.目前我正在SQL数据库中保存所有连接状态(请参阅此文档 ;即使它是"旧"SignalR库的手册,逻辑也是相同的).我也使用Redis背板,因为我的应用程序可以水平扩展.

但是,重新启动我的应用程序时,当前连接不会关闭并将变为孤立状态.之前链接的文章指出:

如果Web服务器停止工作或应用程序重新启动,则不会调用OnDisconnected方法.因此,您的数据存储库可能具有不再有效的连接ID的记录.要清除这些孤立的记录,您可能希望使与应用程序相关的时间范围之外创建的任何连接无效.

这个问题

在"旧"SignalR中有一个ITransportHeartbeat(这个脚本完美实现)但是.NET Core版本没有这样的接口(至少,我找不到它).

我怎么知道连接是否不再存在？我想(或实际需要)清理旧的连接ID.

Answer 1

我想出的解决方案如下。它没有那么优雅，但现在我看不到其他选择。

我更新了数据库中的模型，使其不仅包含 aConnectionId还包含a LastPing（这是一种DateTime类型）。客户端发送KeepAlive消息（自定义消息，不使用 SignalR 保持活动设置）。收到消息（服务器端）后，我用当前时间更新数据库：

var connection = _context.Connection.FirstOrDefault(x => x.Id == Context.ConnectionId);
connection.LastPing = DateTime.UtcNow;

为了清理孤立的连接（不会被 SignalR 的OnDisconnected方法删除），我有一个定期运行的任务（当前在 Hangfire 中），它删除LastPing最近没有更新字段的连接。

Answer 2

在其他答案中@davidfowl 的评论后更新。

带有 SignalR 的 .NET Core 2.1IConnectionHeartbeatFeature可以用来实现与ITransportHeartbeat旧 SignalR 中类似的功能。

下面代码的主要症结是我们维护一个内存列表，用于跟踪数据库中需要更新的连接。这使我们能够在受控的时间间隔内批量执行昂贵的数据库操作。IConnectionHeartbeatFeature.OnHeartbeat()每个连接每秒都会触发一次，因此以该频率访问数据库可能会导致服务器大规模瘫痪。

首先创建一个实体来维护内存中服务器尚未更新的连接列表：

public interface IConnectionCounter
{
    internal ConcurrentDictionary<string, DateTime> Connections { get; }

    public void RecordConnectionLastSeen(string connectionId);
    public void RemoveConnection(string connectionId);
}

/// <summary>
/// Maintains a dictionary of connections that need to be refreshed in the 
/// database
/// </summary>
public class ConnectionCounter : IConnectionCounter
{
    private readonly ConcurrentDictionary<string, DateTime> _connections;
    ConcurrentDictionary<string, DateTime> IConnectionCounter.Connections 
        => _connections;

    public ConnectionCounter()
    {
        _connections = new ConcurrentDictionary<string, DateTime>();
    }

    public void RecordConnectionLastSeen(string connectionId)
    {
        var now = DateTime.UtcNow;
        _connections.AddOrUpdate(
            connectionId, 
            now, (existingConnectionId, oldTime) => now);
    }

    public void RemoveConnection(string connectionId)
    {
        _connections.Remove(connectionId, out _);
    }
}

请注意，这并不是需要更新的所有在线连接的明确列表，因为连接可能分布在多个服务器上。如果您有许多服务器，则可以通过将这些连接存储在分布式内存存储（如 Redis）中来进一步减少负载。

接下来，在集线器中设置 IConnectionCounter，以便对连接进行计数。

public class ChatHub : Hub
{
    private readonly IConnectionCounter _connectionCounter;

    public ChatHub(
        IConnectionCounter connectionCounter)
    {
        _connectionCounter = connectionCounter;
    }

    [AllowAnonymous]
    public override Task OnConnectedAsync()
    {
        var connectionHeartbeat = 
            Context.Features.Get<IConnectionHeartbeatFeature>();

        connectionHeartbeat.OnHeartbeat(connectionId => {
            _connectionCounter.RecordConnectionLastSeen((string)connectionId); 
        }, Context.ConnectionId);

        return base.OnConnectedAsync();
    }
}

现在创建一个服务，该服务获取 IConnectionCounter 中的连接并使用所述连接的状态更新数据库：

public interface IPresenceDatabaseSyncer
{
    public Task UpdateConnectionsOnlineStatus();
}

/// <summary>
/// Handles updating the online status of connections whose connections
/// that need to be updated in the database
/// </summary>
public class PresenceDatabaseSyncer : IPresenceDatabaseSyncer
{
    private readonly MyDbContext _context;
    private readonly IConnectionCounter _connectionCounter;

    public PresenceDatabaseSyncer(
        MyDbContext context, 
        IConnectionCounter connectionCounter)
    {
        _context = context;
        _connectionCounter = connectionCounter;
    }

    public async Task UpdateConnectionsOnlineStatus()
    {
        if (_connectionCounter.Connections.IsEmpty)
            return;

        foreach (var connection in _connectionCounter.Connections)
        {
            var connectionId = connection.Key;
            var lastPing = connection.Value;

            var dbConnection = _context.Connection
                .FirstOrDefault(x => x.ConnectionId == connectionId);

            if (dbConnection != null)
                dbConnection.LastPing = lastPing;

            _connectionCounter.RemoveConnection(connectionId);
        }
    }
}

然后，我使用 HostedService 持续运行上面的数据库同步：

/// <summary>
/// Runs a periodic sync operation to ensure that connections are 
/// recorded as being online correctly in the database
/// </summary>
public class PresenceDatabaseSyncerHostedService : IHostedService, IDisposable
{
    private const int SyncIntervalSeconds = 10;

    private readonly IServiceScopeFactory _serviceScopeFactory;
    private Timer _timer;

    public PresenceDatabaseSyncerHostedService(
        IServiceScopeFactory serviceScopeFactory)
    {
        _serviceScopeFactory = serviceScopeFactory;
    }

    public Task StartAsync(CancellationToken stoppingToken)
    {
        _timer = new Timer(
            DoWork, 
            null, 
            TimeSpan.Zero, 
            TimeSpan.FromSeconds(SyncIntervalSeconds));

        return Task.CompletedTask;
    }

    private async void DoWork(object state)
    {
        using var scope = _serviceScopeFactory.CreateScope();
        var scopedProcessingService = 
            scope.ServiceProvider.GetRequiredService<IPresenceDatabaseSyncer>();

        await scopedProcessingService.UpdateConnectionsOnlineStatus();
    }

    public Task StopAsync(CancellationToken stoppingToken)
    {
        _timer?.Change(Timeout.Infinite, 0);
        return Task.CompletedTask;
    }

    public void Dispose()
    {
        _timer?.Dispose();
    }
}

最后注册这些依赖项和服务：

public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddSingleton<IConnectionCounter, ConnectionCounter>();
        services.AddScoped<IPresenceDatabaseSyncer, PresenceDatabaseSyncer>();

        services.AddHostedService<PresenceDatabaseSyncerHostedService>();
        // ...
    }
    
    // ...
}

当然，仍然存在实际清理数据库中过时连接的问题。我使用另一个 HostedService 来处理这个问题，并将作为练习留给读者。

如果您使用的是 Azure SignalR 服务，那么与按照 @Devator 的答案手动发送 KeepAlive 消息相比，您不需要为消息付费（因为 OnHeartbeat 在内部发生）。

请记住，这个功能并没有得到很好的记录。我已经在生产中使用它几个月了，但我还没有看到使用这种技术的其他解决方案。