如何在Angular前端实现显示Flask API调用进度的进度条？

针对你的需求，以下是几种可行的实现方案，附具体代码和选型建议：

方案1：Server-Sent Events (SSE) （推荐单向进度推送场景） #

SSE是浏览器原生支持的服务器单向推送协议，专门用于后端向前端发送实时更新，比WebSocket轻量，不需要维护双向连接，完全适配进度推送这类单向场景。

Flask后端实现 #

from flask import Flask, Response, stream_with_context
import time

app = Flask(__name__)

@app.route('/long-task')
def long_task():
    def generate_progress():
        # 模拟耗时业务逻辑（如文件处理、第三方API调用）
        for i in range(100):
            time.sleep(0.05)
            # 必须按SSE格式输出：data: 内容\n\n
            yield f'data: {i+1}\n\n'
        # 任务完成后发送结束标记
        yield 'data: done\n\n'

    # 设置响应头为SSE专用类型
    return Response(stream_with_context(generate_progress()), mimetype='text/event-stream')

Angular前端实现 #

用原生EventSource接收SSE消息，绑定到进度条：

import { Component, OnInit } from '@angular/core';

@Component({
  selector: 'app-progress-bar',
  template: `<div class="progress" style="height: 20px; background: #eee; border-radius: 10px;">
               <div class="progress-bar" 
                    [style.width.%]="progress" 
                    style="height: 100%; background: #2196F3; border-radius: 10px;">
               </div>
             </div>`
})
export class ProgressBarComponent implements OnInit {
  progress = 0;

  ngOnInit(): void {
    const eventSource = new EventSource('http://localhost:5000/long-task');
    
    eventSource.onmessage = (event) => {
      if (event.data === 'done') {
        eventSource.close();
        this.progress = 100;
        // 可添加任务完成提示逻辑
        return;
      }
      this.progress = parseInt(event.data);
    };

    eventSource.onerror = (error) => {
      console.error('SSE连接异常:', error);
      eventSource.close();
    };
  }
}

方案2：WebSocket（适合双向交互场景） #

如果你的业务需要前端中途控制任务（如暂停、取消），WebSocket的双向通信能力会更合适。可以用Flask-SocketIO和ngx-socket-io快速实现。

Flask后端实现 #

from flask import Flask
from flask_socketio import SocketIO, emit
import time

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*")

@socketio.on('start_task')
def handle_task_start():
    for i in range(100):
        time.sleep(0.05)
        # 向前端推送进度
        emit('progress_update', {'progress': i+1})
    # 任务完成标记
    emit('progress_update', {'progress': 100, 'done': True})

if __name__ == '__main__':
    socketio.run(app, debug=True)

Angular前端实现 #

先安装依赖：npm install ngx-socket-io

import { Component, OnInit } from '@angular/core';
import { Socket } from 'ngx-socket-io';

@Component({
  selector: 'app-progress-bar',
  template: `<div class="progress" style="height: 20px; background: #eee; border-radius: 10px;">
               <div class="progress-bar" 
                    [style.width.%]="progress" 
                    style="height: 100%; background: #2196F3; border-radius: 10px;">
               </div>
             </div>
             <button (click)="startTask()" style="margin-top: 10px;">启动任务</button>`
})
export class ProgressBarComponent implements OnInit {
  progress = 0;

  constructor(private socket: Socket) {}

  ngOnInit(): void {
    this.socket.fromEvent('progress_update').subscribe((data: any) => {
      this.progress = data.progress;
      if (data.done) {
        // 任务完成逻辑
      }
    });
  }

  startTask(): void {
    this.socket.emit('start_task');
  }
}

方案3：轮询（简单但低效） #

如果不想引入新协议，也可以用定时轮询：后端保存任务进度到内存/数据库，前端每隔一段时间请求进度接口。这种方式实现最简单，但会产生多余HTTP请求，适合进度更新不频繁的场景。

Flask后端 #

from flask import Flask, jsonify
import time
from threading import Thread

app = Flask(__name__)
task_progress = 0

def long_running_task():
    global task_progress
    task_progress = 0
    for i in range(100):
        time.sleep(0.05)
        task_progress = i+1
    task_progress = 100

@app.route('/start-task')
def start_task():
    Thread(target=long_running_task).start()
    return jsonify({'status': '任务已启动'})

@app.route('/get-progress')
def get_progress():
    return jsonify({'progress': task_progress})

Angular前端 #

import { Component, OnInit, OnDestroy } from '@angular/core';
import { HttpClient } from '@angular/common/http';
import { interval, Subscription } from 'rxjs';

@Component({
  selector: 'app-progress-bar',
  template: `<div class="progress" style="height: 20px; background: #eee; border-radius: 10px;">
               <div class="progress-bar" 
                    [style.width.%]="progress" 
                    style="height: 100%; background: #2196F3; border-radius: 10px;">
               </div>
             </div>
             <button (click)="startTask()" style="margin-top: 10px;">启动任务</button>`
})
export class ProgressBarComponent implements OnInit, OnDestroy {
  progress = 0;
  pollSubscription?: Subscription;

  constructor(private http: HttpClient) {}

  ngOnInit(): void {}

  startTask(): void {
    this.http.get('http://localhost:5000/start-task').subscribe(() => {
      // 每隔500ms请求一次进度
      this.pollSubscription = interval(500).subscribe(() => {
        this.http.get('http://localhost:5000/get-progress').subscribe((data: any) => {
          this.progress = data.progress;
          if (this.progress === 100) {
            this.pollSubscription?.unsubscribe();
          }
        });
      });
    });
  }

  ngOnDestroy(): void {
    this.pollSubscription?.unsubscribe();
  }
}

选型建议 #

• 仅需单向进度推送：优先选SSE，实现简单、轻量，无额外依赖，浏览器原生支持。
• 需要双向交互：选WebSocket，适合前端控制任务的场景。
• 快速原型/简单场景：选轮询，无需学习新协议，但注意性能损耗。

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