基于Flutter Bloc实现SignalR并发/并行请求的方法

核心需求 #

• 在Flutter Bloc中发起并发/并行的SignalR请求，无需等待前一个请求完成
• 在请求处理过程中，实时更新并展示所有请求的结果，直到满足特定条件

现有实现的问题 #

当前代码通过for循环串行处理请求，每个请求必须等上一个完成才会执行，无法实现并发：

on<eventName>(
        (event, emit) async {      
          for (final element in event.files) {
            await establishSignalRConnection(uploadChannelId, element.name);
            final DocumentUpload uploadAttachment =
                await _Repository.upload();          
            emit(uploadedState(
                 emittedStateProperty
                ));
          }
        },
      );

解决方案 #

1. 基于Future.wait实现基础并发 #

将所有上传请求包装为Future，通过Future.wait让它们并行执行，单个请求完成后立即更新状态，实现实时展示结果：

on<eventName>(
  (event, emit) async {
    // 存储所有已完成的上传结果
    final List<DocumentUpload> uploadResults = [];
    
    // 生成所有并发上传任务
    final uploadFutures = event.files.map((element) async {
      await establishSignalRConnection(uploadChannelId, element.name);
      final uploadResult = await _Repository.upload();
      
      // 单个请求完成后，更新结果列表并发射最新状态
      uploadResults.add(uploadResult);
      emit(UploadedState(results: List.from(uploadResults)));
      
      return uploadResult;
    }).toList();

    // 等待所有上传任务完成（可根据业务条件调整是否需要等待全部完成）
    await Future.wait(uploadFutures);
    
    // 所有请求完成后发射最终状态（可选）
    emit(AllUploadsCompletedState(results: uploadResults));
  },
);

2. 控制并发数量（可选） #

如果需要限制同时发起的请求数量（避免资源过载），可以使用Semaphore做并发控制：

// 限制同时最多3个并发请求
final uploadSemaphore = Semaphore(3);

on<eventName>(
  (event, emit) async {
    final List<DocumentUpload> uploadResults = [];
    final uploadFutures = event.files.map((element) async {
      await uploadSemaphore.acquire();
      try {
        await establishSignalRConnection(uploadChannelId, element.name);
        final uploadResult = await _Repository.upload();
        
        uploadResults.add(uploadResult);
        emit(UploadedState(results: List.from(uploadResults)));
      } finally {
        // 释放信号量，允许下一个请求执行
        uploadSemaphore.release();
      }
      return uploadResult;
    }).toList();

    await Future.wait(uploadFutures);
    emit(AllUploadsCompletedState(results: uploadResults));
  },
  concurrent: true, // 允许该事件的多个实例并发处理
);

3. 实时高频更新（每毫秒展示） #

如果需要严格按每毫秒频率更新状态，可以结合Stream.periodic定时触发状态发射：

on<eventName>(
  (event, emit) async {
    final List<DocumentUpload> uploadResults = [];
    final uploadFutures = event.files.map((element) async {
      await establishSignalRConnection(uploadChannelId, element.name);
      final uploadResult = await _Repository.upload();
      uploadResults.add(uploadResult);
      return uploadResult;
    }).toList();

    // 每毫秒发射一次当前的结果状态
    final updateStream = Stream.periodic(const Duration(milliseconds: 1), (_) {
      emit(UploadedState(results: List.from(uploadResults)));
    });

    // 等待所有上传完成后关闭定时更新
    await Future.wait(uploadFutures);
    updateStream.drain();
    emit(AllUploadsCompletedState(results: uploadResults));
  },
);

关键说明 #

• Bloc的emit方法是线程安全的，多次调用会自动处理状态的增量更新，UI可直接基于最新状态渲染所有请求结果
• Future.wait仅等待所有任务完成，但单个任务完成时即可触发状态更新，无需等全部结束
• 开启concurrent: true可允许同一事件的多个实例同时处理，适配多批次并发上传场景

内容的提问来源于stack exchange，提问作者J.Dhinesh