Angular 5文件上传组件：订阅提前完成及错误列表展示问题

你遇到的核心问题很明确：Observable.from(files)是同步发射所有文件对象的——哪怕你在订阅逻辑里调用了异步的上传服务，subscribe的complete回调也会在所有文件的同步校验代码执行完就立刻触发，完全等不及异步上传请求完成，这就是你看到"subscription complete called before completion"的原因。

要解决这个问题，我们需要把每个文件的完整处理流程（校验+上传）包装成独立的Observable，然后用RxJS的合并操作符等待所有异步操作完成后再处理错误列表。下面是具体的解决方案：

步骤1：重构单个文件的处理逻辑 #

先把单个文件的校验和上传逻辑封装成一个返回Observable的函数，让每个文件的异步操作都能被RxJS追踪：

// 假设组件已定义MAX_SIZE、validFileTypes，且注入了uploadService
private errors: string[] = [];

private processFile(file: File): Observable<void> {
  // 先执行同步校验
  if (file.size > this.MAX_SIZE) {
    this.errors.push(`${file.name} is too large.`);
    return Observable.of(void 0); // 返回已完成的Observable
  } 
  if (this.validFileTypes.indexOf(file.type) < 0) {
    this.errors.push(`${file.name} is not a valid image file.`);
    return Observable.of(void 0);
  }

  // 校验通过则调用上传服务，同时捕获上传错误
  return this.uploadService.upload(file).pipe(
    catchError((error) => {
      this.errors.push(`${file.name} upload failed: ${error.message}`);
      return Observable.of(void 0); // 捕获错误，避免整个流中断
    })
  );
}

步骤2：合并所有文件的Observable并等待完成 #

接下来把所有文件转成Observable数组，用forkJoin（并行处理）或者concat（顺序处理）等待所有操作完成后再展示错误列表：

并行处理所有文件（推荐，上传速度更快） #

import { forkJoin } from 'rxjs/observable/forkJoin';
import { Observable } from 'rxjs/Observable';
import 'rxjs/add/observable/of';
import 'rxjs/add/operator/catch';

// 将文件数组转为Observable数组
const fileProcessObservables = Array.from(files).map(file => this.processFile(file));

forkJoin(fileProcessObservables).subscribe(
  () => {
    // 所有文件的校验、上传操作都已完成
    console.log('All files processed');
    // 在这里展示错误列表
    this.displayErrors(this.errors);
  },
  (unexpectedError) => {
    // 仅forkJoin自身异常时触发，上传错误已在processFile中捕获
    console.error('Unexpected error:', unexpectedError);
  }
);

顺序处理文件（适合需按顺序上传的场景） #

如果业务要求文件必须按顺序上传，用concat代替forkJoin即可：

import { concat } from 'rxjs/observable/concat';

concat(...fileProcessObservables).subscribe(
  () => {}, // 单个文件处理完成的回调，可选
  () => {}, // 错误回调，可选
  () => {
    // 所有文件处理完成
    this.displayErrors(this.errors);
  }
);

关键说明 #

• forkJoin会等待所有传入的Observable都完成后才发射结果，确保我们能在所有文件处理完成后再展示错误列表。
• 在processFile中用catchError捕获上传错误，单个文件上传失败不会导致整个流中断，其他文件仍能正常处理。
• 把errors定义为组件的成员变量，避免闭包导致的变量访问问题。

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