基于ASP.NET Core Web API与Angular的批量数据处理超时解决方案咨询

一、核心架构设计 #

把耗时的LDAP查询+数据库更新逻辑从API请求线程剥离，放到后台任务队列执行，同时用SignalR建立客户端与服务器的实时连接，任务进度更新、完成或失败时推送给客户端。这样既避免API请求超时，用户跳转页面后只要重新连接SignalR，就能继续接收任务状态通知。

二、具体实现步骤 #

1. 配置ASP.NET Core后台任务队列 #

• 实现线程安全的后台任务队列接口，用Channel存储待执行任务（比普通Queue更适合异步场景）：

  public interface IBackgroundTaskQueue
  {
      ValueTask QueueBackgroundWorkItemAsync(Func<CancellationToken, ValueTask> workItem);
      ValueTask<Func<CancellationToken, ValueTask>> DequeueAsync(CancellationToken cancellationToken);
  }
  
  public class BackgroundTaskQueue : IBackgroundTaskQueue
  {
      private readonly Channel<Func<CancellationToken, ValueTask>> _queue;
  
      public BackgroundTaskQueue(int capacity)
      {
          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);
      }
  }
  

• 编写后台任务处理服务，轮询队列执行任务：

  public class TaskProcessingService : BackgroundService
  {
      private readonly IBackgroundTaskQueue _taskQueue;
      private readonly ILogger<TaskProcessingService> _logger;
      private readonly IHubContext<NotificationHub> _hubContext;
  
      public TaskProcessingService(IBackgroundTaskQueue taskQueue, ILogger<TaskProcessingService> logger, IHubContext<NotificationHub> hubContext)
      {
          _taskQueue = taskQueue;
          _logger = logger;
          _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, "后台任务执行失败");
              }
          }
      }
  }
  

• 在Program.cs注册服务：

  builder.Services.AddSingleton<IBackgroundTaskQueue>(_ => new BackgroundTaskQueue(100));
  builder.Services.AddHostedService<TaskProcessingService>();
  

2. 集成SignalR实现实时通知 #

• 创建SignalR Hub，支持按用户分组推送通知：

  public class NotificationHub : Hub
  {
      public async Task RegisterClient(string userId)
      {
          await Groups.AddToGroupAsync(Context.ConnectionId, userId);
      }
  }
  

• 在Program.cs配置SignalR路由：

  app.MapHub<NotificationHub>("/notificationHub");
  

3. API接口改造 #

• 原API不再直接执行耗时逻辑，而是将任务加入后台队列，立即返回任务ID供客户端跟踪：

  [ApiController]
  [Route("api/ldap-sync")]
  public class LdapSyncController : ControllerBase
  {
      private readonly IBackgroundTaskQueue _taskQueue;
      private readonly IHubContext<NotificationHub> _hubContext;
  
      public LdapSyncController(IBackgroundTaskQueue taskQueue, IHubContext<NotificationHub> hubContext)
      {
          _taskQueue = taskQueue;
          _hubContext = hubContext;
      }
  
      [HttpPost("start")]
      public async Task<IActionResult> StartSync([FromBody] SyncRequest request)
      {
          var taskId = Guid.NewGuid().ToString();
          await _taskQueue.QueueBackgroundWorkItemAsync(async token =>
          {
              try
              {
                  // 模拟进度推送（可选）
                  await _hubContext.Clients.Group(request.UserId).SendAsync("SyncProgress", new { TaskId = taskId, Progress = 30 });
                  // 执行LDAP批量查询
                  var ldapRecords = await QueryLdapBatch(token);
                  await _hubContext.Clients.Group(request.UserId).SendAsync("SyncProgress", new { TaskId = taskId, Progress = 60 });
                  // 批量更新数据库
                  await BatchUpdateDatabase(ldapRecords, token);
                  await _hubContext.Clients.Group(request.UserId).SendAsync("SyncCompleted", new { TaskId = taskId, Success = true, Message = "同步完成" });
              }
              catch (Exception ex)
              {
                  await _hubContext.Clients.Group(request.UserId).SendAsync("SyncFailed", new { TaskId = taskId, Message = ex.Message });
              }
          });
          return Ok(new { TaskId = taskId });
      }
  }
  
  public class SyncRequest
  {
      public string UserId { get; set; } = string.Empty;
      // 其他同步参数
  }
  

4. Angular客户端实现 #

• 安装SignalR依赖：npm install @microsoft/signalr
• 封装SignalR服务，管理连接与通知监听：

  import { Injectable } from '@angular/core';
  import { HubConnection, HubConnectionBuilder } from '@microsoft/signalr';
  import { BehaviorSubject } from 'rxjs';
  
  @Injectable({ providedIn: 'root' })
  export class SignalRService {
      private hubConnection!: HubConnection;
      public syncProgress$ = new BehaviorSubject<{ taskId: string; progress: number } | null>(null);
      public syncStatus$ = new BehaviorSubject<{ taskId: string; success: boolean; message: string } | null>(null);
  
      startConnection(userId: string) {
          this.hubConnection = new HubConnectionBuilder()
              .withUrl('/notificationHub')
              .withAutomaticReconnect()
              .build();
  
          this.hubConnection.start().then(() => {
              this.hubConnection.invoke('RegisterClient', userId);
          }).catch(err => console.error('SignalR连接失败:', err));
  
          this.hubConnection.on('SyncProgress', (data) => {
              this.syncProgress$.next(data);
          });
  
          this.hubConnection.on('SyncCompleted', (data) => {
              this.syncStatus$.next({ ...data, success: true });
          });
  
          this.hubConnection.on('SyncFailed', (data) => {
              this.syncStatus$.next({ ...data, success: false });
          });
      }
  
      stopConnection() {
          this.hubConnection.stop().catch(err => console.error('SignalR断开失败:', err));
      }
  }
  

• 在组件中使用：启动同步后，即使跳转页面，可在根组件或路由守卫中重新初始化SignalR连接，继续接收任务通知。

三、关键注意事项 #

• 任务持久化：若服务器重启，未完成的任务会丢失，可将任务信息存入数据库或Redis，重启后重新加载执行。
• 并发控制：后台队列容量需根据服务器CPU、内存配置调整，避免同时执行过多任务导致资源耗尽。
• LDAP与数据库优化：LDAP用批量查询减少请求次数；数据库用AddRange/UpdateRange批量操作，或直接执行SQL批量语句，提升效率。
• 客户端重连：利用SignalR自动重连机制，结合用户ID分组，确保用户刷新/跳转页面后能重新接收对应任务的通知。

内容的提问来源于stack exchange，提问作者S.Siva