node.js - 定期发送 API 的进度更新并保存进度

在我的 NestJS 应用中，我正在尝试处理一些可能需要 10-20 分钟才能完成的任务。因此，用户点击一个按钮，并调用一个 API，该 API 开始执行一些需要这个时间量的处理。

为了更清楚地说明，想象这样一个情况：有一个网络应用程序，其中有一个过程需要一些时间。最终你会在前端向用户显示进度更新。所以用户可能会看到“正在生成您的文档摘要”，然后是“正在做一些相关工作”等等。

在此过程中，我希望进度也能被保存，这样如果用户导航离开，他们仍然可以回来从进度停下的地方继续（即界面会被更新）。一旦所有内容完成，它将继续前进。

我希望当用户调用该 API 时，它不会阻碍进程或被阻塞。我曾考虑过使用服务器发送事件（SSE）来实现这一点，但如果用户离开页面，SSE 就会断开连接。那么，我应该如何实现这一点呢？在一个我正在从 OpenAI 生成响应的 API 中，我正在使用 SSE 来获得更实时的体验（这是我的代码的简化片段）：

@Sse(':id/generate-answer')
async answerWithMagic(
    @Param('id') id,
) {
    const messages = this.aiService.resolvePrompt();
    const stream = new Subject();
    this.aiService
        .generateCompletion({ messages, listenTokens: true, isJsonParsable: false })
        .pipe(
            takeWhile((data) => {
                if (data.eventType === LLM_END_EVENT) {
                    stream.next({
                        eventType: data.eventType,
                        data: { token: data.token },
                    });
                    return false;
                } else {
                    return true;
                }
            }),
        )
        .subscribe((data) => {
            stream.next({ eventType: data.eventType, data: { token: data.token } });
        });

    return stream;
}

我如何在这里保存进度？反复调用数据库效率不高，所以我考虑使用 Redis 在那里存储进度，但我不确定在这方面应该采取哪种方向。

答案＃1

我认为你的方法还可以。

这个想法是将长任务在后台运行，并使用 Redis 来跟踪进度。对于任务计算，你可以使用 BullMQ（基于 Redis 的队列），它被 NestJS 直接支持：https://docs.nestjs.com/techniques/queues 。

所以，从技术上讲，你需要创建一个 RedisService，它将包含用于设置和获取任务进度的方法：

@Injectable()
export class RedisService {
  private redis: Redis;

  constructor() {
    this.redis = new Redis({
      host: process.env.REDIS_HOST || 'localhost',
      port: parseInt(process.env.REDIS_PORT, 10) || 6379,
    });
  }

  async setProgress(taskId: string, progress: any) {
    await this.redis.set(`progress:${taskId}`, JSON.stringify(progress), 'EX', 3600); // TTL 1 hour
  }

  async getProgress(taskId: string) {
    const progress = await this.redis.get(`progress:${taskId}`);
    return progress ? JSON.parse(progress) : null;
  }
}

然后，你必须将 RedisService 注入到处理任务以存储进度的服务中。

之后，你只需创建另一个端点来获取任务更新：

  @Post(':id/generate-answer')
  async startTask(@Param('id') taskId: string) {
    // Run task with BullMQ
  }

  @Get(':id/progress')
  async getProgress(@Param('id') taskId: string) {
    const progress = await this.redisService.getProgress(taskId);
    return progress ?? { progress: 0, message: 'Not started' };
  }

前端可以每隔几秒轮询 /tasks/:id/progress（或者如果他们停留，则重新连接 SSE）。