Spring Boot+Angular应用中如何实现批量处理进度的用户通知？

问题背景 #

我的应用采用Spring Boot后端与Angular前端架构，前端通过REST接口调用后端服务。需要实现一个耗时数分钟的批量处理流程，包含调用大量外部API、数据处理及数据库写入，希望前端能实时展示进度消息，比如：

正在从外部服务请求索引...已接收，需查询100条数据。
正在发起100次Widget数据API调用...已完成100次请求，95次成功，5次终止。
正在处理结果...完成。
正在写入数据库...完成。

核心疑问：

• 是否需要让Angular前端编排所有流程，发起数十次后端请求？
• Spring Boot是否支持创建可在批量处理过程中多次返回数据的端点，直接向前端推送进度？

解决方案 #

不需要前端编排流程，Spring Boot支持通过流式响应（SSE）或WebSocket实现服务器主动向前端推送进度，以下是具体实现方案：

方案一：Server-Sent Events（SSE）单向推送 #

这是最适合你场景的方案，基于HTTP协议实现服务器单向推送进度，无需前端频繁请求，代码简洁且职责划分清晰。

后端Spring Boot实现

通过SseEmitter创建流式接口，异步执行批量处理并实时推送进度：

@RestController
@RequestMapping("/batch")
public class BatchProcessController {

    @GetMapping("/start")
    public SseEmitter startBatchProcess() {
        // 设置超时时间为最大值，避免中途断开
        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);

        // 异步执行批量处理，不阻塞请求线程
        CompletableFuture.runAsync(() -> {
            try {
                // 步骤1：请求外部索引
                emitter.send(SseEmitter.event().name("progress").data("正在从外部服务请求索引..."));
                int totalItems = 100;
                emitter.send(SseEmitter.event().name("progress").data("正在从外部服务请求索引...已接收，需查询" + totalItems + "条数据。"));

                // 步骤2：批量调用外部API
                emitter.send(SseEmitter.event().name("progress").data("正在发起" + totalItems + "次Widget数据API调用..."));
                int successCount = 95;
                int failCount = 5;
                // 模拟分批推送进度（可选，根据实际需求调整）
                for (int i = 1; i <= totalItems; i++) {
                    Thread.sleep(50); // 模拟单次API调用耗时
                    if (i % 10 == 0) {
                        emitter.send(SseEmitter.event().name("progress").data("正在发起" + totalItems + "次Widget数据API调用...已完成" + i + "次请求"));
                    }
                }
                emitter.send(SseEmitter.event().name("progress").data("正在发起" + totalItems + "次Widget数据API调用...已完成100次请求，" + successCount + "次成功，" + failCount + "次终止。"));

                // 步骤3：处理结果
                emitter.send(SseEmitter.event().name("progress").data("正在处理结果..."));
                Thread.sleep(1000);
                emitter.send(SseEmitter.event().name("progress").data("正在处理结果...完成。"));

                // 步骤4：写入数据库
                emitter.send(SseEmitter.event().name("progress").data("正在写入数据库..."));
                Thread.sleep(1500);
                emitter.send(SseEmitter.event().name("progress").data("正在写入数据库...完成。"));

                // 发送完成事件并关闭连接
                emitter.send(SseEmitter.event().name("complete").data("批量处理已全部完成"));
                emitter.complete();
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        });

        return emitter;
    }
}

前端Angular实现

使用EventSource监听SSE流，实时更新页面进度：

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

@Component({
  selector: 'app-batch-progress',
  template: `
    <button (click)="startBatch()">启动批量处理</button>
    <div class="progress-container">
      <p *ngFor="let msg of progressMessages" class="progress-msg">{{ msg }}</p>
    </div>
  `,
  styles: [`
    .progress-container { margin-top: 20px; padding: 10px; border: 1px solid #eee; border-radius: 4px; }
    .progress-msg { margin: 8px 0; font-size: 14px; }
  `]
})
export class BatchProgressComponent {
  progressMessages: string[] = [];
  eventSource: EventSource | null = null;

  startBatch() {
    this.progressMessages = [];
    // 建立SSE连接
    this.eventSource = new EventSource('/batch/start');

    // 监听进度事件
    this.eventSource.addEventListener('progress', (event) => {
      const message = (event as MessageEvent).data;
      this.progressMessages.push(message);
    });

    // 监听完成事件
    this.eventSource.addEventListener('complete', (event) => {
      const message = (event as MessageEvent).data;
      this.progressMessages.push(message);
      this.eventSource?.close();
    });

    // 监听错误事件
    this.eventSource.onerror = (error) => {
      this.progressMessages.push(`批量处理出错：${error.message}`);
      this.eventSource?.close();
    };
  }
}

方案二：WebSocket双向通信 #

如果需要支持前端中途取消批量处理等双向交互场景，可以使用WebSocket：

后端Spring Boot实现

配置WebSocket并处理进度推送：

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(new BatchProgressHandler(), "/batch-ws").setAllowedOrigins("*");
    }
}

public class BatchProgressHandler extends TextWebSocketHandler {
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        // 连接建立后异步执行批量处理
        CompletableFuture.runAsync(() -> {
            try {
                // 推送各步骤进度，逻辑同SSE示例
                session.sendMessage(new TextMessage("正在从外部服务请求索引..."));
                Thread.sleep(1000);
                session.sendMessage(new TextMessage("正在从外部服务请求索引...已接收，需查询100条数据。"));
                // ...后续步骤推送
                session.sendMessage(new TextMessage("批量处理已全部完成"));
                session.close();
            } catch (Exception e) {
                try {
                    session.sendMessage(new TextMessage(`处理出错：${e.getMessage()}`));
                    session.close();
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }
        });
    }
}

前端Angular实现

使用WebSocket客户端接收进度：

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

@Component({
  selector: 'app-batch-ws',
  template: `
    <button (click)="startBatch()">启动批量处理</button>
    <div class="progress-container">
      <p *ngFor="let msg of progressMessages" class="progress-msg">{{ msg }}</p>
    </div>
  `
})
export class BatchWsComponent {
  progressMessages: string[] = [];
  webSocket: WebSocket | null = null;

  startBatch() {
    this.progressMessages = [];
    this.webSocket = new WebSocket('ws://localhost:8080/batch-ws');

    this.webSocket.onmessage = (event) => {
      this.progressMessages.push(event.data);
    };

    this.webSocket.onerror = (error) => {
      this.progressMessages.push(`WebSocket连接出错：${error.message}`);
      this.webSocket?.close();
    };

    this.webSocket.onclose = () => {
      this.progressMessages.push("连接已关闭");
    };
  }
}

方案三：前端轮询（传统兼容方案） #

如果不需要实时推送，也可以用前端定期调用后端接口查询进度，适合对兼容性要求极高的场景：

后端实现

保存批量处理进度状态，提供查询接口：

@RestController
@RequestMapping("/batch")
public class BatchPollController {
    private final Map<String, List<String>> progressMap = new ConcurrentHashMap<>();

    @PostMapping("/start-poll")
    public ResponseEntity<String> startBatchPoll(@RequestBody String requestId) {
        progressMap.put(requestId, new ArrayList<>());
        CompletableFuture.runAsync(() -> {
            List<String> progress = progressMap.get(requestId);
            try {
                progress.add("正在从外部服务请求索引...");
                Thread.sleep(1000);
                progress.add("正在从外部服务请求索引...已接收，需查询100条数据。");
                // ...后续步骤更新进度
                progress.add("批量处理已全部完成");
            } catch (Exception e) {
                progress.add(`处理出错：${e.getMessage()}`);
            }
        });
        return ResponseEntity.ok(requestId);
    }

    @GetMapping("/progress/{requestId}")
    public ResponseEntity<List<String>> getProgress(@PathVariable String requestId) {
        return ResponseEntity.ok(progressMap.getOrDefault(requestId, new ArrayList<>()));
    }
}

前端实现

用定时器定期查询进度：

import { Component } from '@angular/core';
import { HttpClient } from '@angular/common/http';

@Component({
  selector: 'app-batch-poll',
  template: `
    <button (click)="startBatch()">启动批量处理</button>
    <div class="progress-container">
      <p *ngFor="let msg of progressMessages" class="progress-msg">{{ msg }}</p>
    </div>
  `
})
export class BatchPollComponent {
  progressMessages: string[] = [];
  pollInterval: any;
  requestId: string = Math.random().toString(36).substring(2);

  constructor(private http: HttpClient) {}

  startBatch() {
    this.progressMessages = [];
    // 启动批量处理
    this.http.post('/batch/start-poll', this.requestId, { responseType: 'text' }).subscribe(() => {
      // 每秒查询一次进度
      this.pollInterval = setInterval(() => {
        this.http.get<string[]>(`/batch/progress/${this.requestId}`).subscribe(messages => {
          this.progressMessages = messages;
          // 完成或出错时停止轮询
          if (messages.includes("批量处理已全部完成") || messages.some(msg => msg.startsWith("处理出错"))) {
            clearInterval(this.pollInterval);
          }
        });
      }, 1000);
    });
  }
}

方案对比 #

针对你的疑问：

1. 不需要前端编排流程：后端可以独立完成批量处理并推送进度，职责划分更清晰。
2. Spring Boot支持多次返回数据的端点：通过SSE或WebSocket即可实现，其中SSE是最匹配你需求的方案。

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