Blazor应用中用SignalR调度后台视频导出任务并通知发起用户的问题

针对你提到的「从SignalR Hub推送后台导出任务并完成后通知用户」的问题，核心是避免直接传递Hub实例或ConnectionId到任务回调，而是通过依赖注入获取全局的IHubContext，结合连接状态检测来实现可靠通知。以下是具体实现方案：

1. 核心思路 #

• 后台任务队列使用独立的IBackgroundTaskQueue实现，任务参数包含用户的ConnectionId和导出信息
• 后台服务（BackgroundService）注入IHubContext<ExportHub>，而非从Hub传递实例，避免Hub销毁导致的上下文失效
• 通过SignalR双向调用检测用户是否在线：若用户断开连接，直接取消导出并清理文件；若在线则完成导出后推送结果

2. 代码实现步骤 #

2.1 定义后台任务队列 #

先实现一个通用的后台任务队列，用于接收导出请求：

public interface IBackgroundTaskQueue
{
    ValueTask QueueBackgroundWorkItemAsync(Func<CancellationToken, ValueTask> workItem);
}

public class BackgroundTaskQueue : IBackgroundTaskQueue
{
    private readonly Channel<Func<CancellationToken, ValueTask>> _queue;

    public BackgroundTaskQueue(int capacity = 100)
    {
        var options = new BoundedChannelOptions(capacity)
        {
            FullMode = BoundedChannelFullMode.Wait
        };
        _queue = Channel.CreateBounded<Func<CancellationToken, ValueTask>>(options);
    }

    public async ValueTask QueueBackgroundWorkItemAsync(Func<CancellationToken, ValueTask> workItem)
    {
        if (workItem == null)
            throw new ArgumentNullException(nameof(workItem));

        await _queue.Writer.WriteAsync(workItem);
    }

    public async ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(CancellationToken cancellationToken)
    {
        return await _queue.Reader.ReadAsync(cancellationToken);
    }
}

2.2 实现导出后台服务 #

后台服务负责处理队列中的任务，通过IHubContext和用户ConnectionId定向推送消息：

public class ExportBackgroundService : BackgroundService
{
    private readonly ILogger<ExportBackgroundService> _logger;
    private readonly IBackgroundTaskQueue _taskQueue;
    private readonly IHubContext<ExportHub> _hubContext;

    public ExportBackgroundService(ILogger<ExportBackgroundService> logger, 
                                   IBackgroundTaskQueue taskQueue, 
                                   IHubContext<ExportHub> hubContext)
    {
        _logger = logger;
        _taskQueue = taskQueue;
        _hubContext = hubContext;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("导出后台服务已启动");

        while (!stoppingToken.IsCancellationRequested)
        {
            var workItem = await _taskQueue.DequeueAsync(stoppingToken);

            try
            {
                await workItem(stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "处理导出任务时出错");
            }
        }

        _logger.LogInformation("导出后台服务已停止");
    }
}

2.3 实现SignalR导出Hub #

Hub接收用户的导出请求，将任务加入队列，并返回任务ID给前端：

public class ExportHub : Hub
{
    private readonly IBackgroundTaskQueue _taskQueue;

    public ExportHub(IBackgroundTaskQueue taskQueue)
    {
        _taskQueue = taskQueue;
    }

    public async Task StartExport(string videoId, List<Bookmark> bookmarks)
    {
        var connectionId = Context.ConnectionId;
        var taskId = Guid.NewGuid().ToString();

        // 将导出任务加入队列，携带用户连接ID和任务参数
        await _taskQueue.QueueBackgroundWorkItemAsync(async token =>
        {
            // 检测用户是否在线：调用前端方法验证连接
            bool isUserOnline = false;
            try
            {
                isUserOnline = await _hubContext.Clients.Client(connectionId)
                    .InvokeAsync<bool>("IsUserOnline", token);
            }
            catch
            {
                // 调用失败说明用户已断开连接
                isUserOnline = false;
            }

            if (!isUserOnline)
            {
                await _hubContext.Clients.Client(connectionId)
                    .SendAsync("ExportCancelled", taskId, "用户已断开连接");
                return;
            }

            try
            {
                // 执行实际的视频裁剪导出逻辑
                using var exportedStream = await VideoExporter.ClipVideoAsync(videoId, bookmarks, token);
                
                // 通知前端导出完成，传递文件流
                await _hubContext.Clients.Client(connectionId)
                    .SendAsync("ExportCompleted", taskId, exportedStream);

                // 立即清理导出文件（符合需求：不保存文件）
                await CleanupTemporaryFile(exportedStream);
            }
            catch (OperationCanceledException)
            {
                await _hubContext.Clients.Client(connectionId)
                    .SendAsync("ExportCancelled", taskId, "导出任务已取消");
            }
            catch (Exception ex)
            {
                await _hubContext.Clients.Client(connectionId)
                    .SendAsync("ExportFailed", taskId, ex.Message);
            }
        });

        // 立即告知前端任务已启动
        await Clients.Caller.SendAsync("ExportStarted", taskId);
    }
}

2.4 Blazor前端处理 #

在Blazor组件中连接SignalR，接收后台推送的事件，并处理导出结果：

@page "/video/{VideoId}"
@inject NavigationManager NavigationManager
@implements IAsyncDisposable

<h3>视频书签导出</h3>

@if (_hubConnection?.State == HubConnectionState.Connected)
{
    <button @onclick="TriggerExport" disabled="@isExporting">
        @(isExporting ? "导出中..." : "开始导出选中片段")
    </button>
    <p class="mt-2">状态: @exportStatus</p>
    @if (!string.IsNullOrEmpty(downloadUrl))
    {
        <a href="@downloadUrl" download="exported-video.mp4" class="btn btn-success mt-2">
            下载导出视频
        </a>
    }
}
else
{
    <p>正在连接导出服务...</p>
}

@code {
    [Parameter] public string VideoId { get; set; }
    private HubConnection _hubConnection;
    private bool isExporting;
    private string exportStatus = "等待操作";
    private string downloadUrl;
    private List<Bookmark> selectedBookmarks = new(); // 假设已获取用户选择的书签

    protected override async Task OnInitializedAsync()
    {
        _hubConnection = new HubConnectionBuilder()
            .WithUrl(NavigationManager.ToAbsoluteUri("/exportHub"))
            .Build();

        // 注册接收后台推送的事件
        _hubConnection.On<string>(nameof(ExportStarted), taskId =>
        {
            isExporting = true;
            exportStatus = $"任务 {taskId} 已启动，正在处理...";
            StateHasChanged();
        });

        _hubConnection.On<string, Stream>(nameof(ExportCompleted), async (taskId, fileStream) =>
        {
            isExporting = false;
            exportStatus = $"任务 {taskId} 完成";
            
            // 将文件流转换为前端可下载的Base64 URL
            var bytes = new byte[fileStream.Length];
            await fileStream.ReadAsync(bytes, 0, bytes.Length);
            downloadUrl = $"data:video/mp4;base64,{Convert.ToBase64String(bytes)}";
            
            StateHasChanged();
        });

        _hubConnection.On<string, string>(nameof(ExportCancelled), (taskId, reason) =>
        {
            isExporting = false;
            exportStatus = $"任务 {taskId} 已取消: {reason}";
            StateHasChanged();
        });

        _hubConnection.On<string, string>(nameof(ExportFailed), (taskId, error) =>
        {
            isExporting = false;
            exportStatus = $"任务 {taskId} 失败: {error}";
            StateHasChanged();
        });

        // 注册供后台任务调用的在线验证方法
        _hubConnection.On<bool>(nameof(IsUserOnline), () => Task.FromResult(true));

        await _hubConnection.StartAsync();
    }

    private async Task TriggerExport()
    {
        await _hubConnection.SendAsync(nameof(StartExport), VideoId, selectedBookmarks);
    }

    public async ValueTask DisposeAsync()
    {
        if (_hubConnection != null)
        {
            await _hubConnection.DisposeAsync();
        }
    }

    // 辅助类：书签模型（根据实际需求调整）
    public class Bookmark
    {
        public TimeSpan StartTime { get; set; }
        public TimeSpan EndTime { get; set; }
    }
}

2.5 注册服务到DI容器 #

在Program.cs中注册队列、后台服务和SignalR：

builder.Services.AddSingleton<IBackgroundTaskQueue, BackgroundTaskQueue>();
builder.Services.AddHostedService<ExportBackgroundService>();
builder.Services.AddSignalR();

// ...

app.MapHub<ExportHub>("/exportHub");

3. 关键注意事项 #

• 避免传递Hub实例：后台服务通过DI注入IHubContext，它是全局单例，不会随单个Hub实例销毁而失效
• 连接状态检测：通过调用前端的IsUserOnline方法验证用户是否在线，比单纯依赖ConnectionId更可靠（ConnectionId可能在重连时变化）
• 任务取消：后台任务使用CancellationToken，配合前端断开时的Hub销毁事件，可及时取消导出并清理资源
• 文件清理：导出完成后立即删除临时文件，严格遵循「不保存文件」的需求

内容的提问来源于stack exchange，提问作者Expressingx