NestJS中实现同步队列解决Dropbox请求频率超限问题

队列未生效的核心原因 #

你之前接入Bull后未实现限流，90%以上的情况是没有配置队列的限流规则，仅把任务放到队列但未限制单位时间的执行数量，队列会尽可能多地并发执行任务，效果和直接同步调用没有差异。

正确实现步骤 #

1. 确认依赖安装 #

确保已安装匹配你NestJS版本的队列依赖：

npm install @nestjs/bull bull

2. 配置队列限流规则 #

Bull的limiter配置是控制请求频率的核心，可直接匹配Dropbox的限流阈值（Dropbox普通开发者账号一般限制每分钟100~300次请求，可根据自己的账号权限调整参数）。
首先在模块中注册队列时添加限流配置：

// 对应业务模块的module.ts
import { BullModule } from '@nestjs/bull';

@Module({
  imports: [
    BullModule.registerQueue({
      name: 'dropbox-upload',
      limiter: {
        max: 5, // 单位时间内最多执行5个任务
        duration: 1000, // 单位时间为1秒，即每秒最多处理5个上传请求
        bounceBack: true // 触发限流时将任务放回队列尾部，延迟后重试，不丢弃任务
      }
    })
  ]
})
export class UploadModule {}

3. 实现队列消费逻辑 #

配置消费端的并发数，避免单个进程同时执行过多任务：

// dropbox.processor.ts
import { Process, Processor } from '@nestjs/bull';
import { Job } from 'bull';
import * as fs from 'fs/promises';

@Processor('dropbox-upload')
export class DropboxUploadProcessor {
  constructor(private dropboxService: DropboxService) {}

  // 限制单个进程同时最多处理2个上传任务
  @Process({ concurrency: 2 })
  async handleUpload(job: Job<{ localPath: string; dropboxPath: string }>) {
    // 执行Dropbox上传逻辑
    await this.dropboxService.upload(job.data.localPath, job.data.dropboxPath);
    // 上传成功后删除本地文件
    await fs.unlink(job.data.localPath);
  }
}

4. 接口层仅提交任务到队列，不执行同步上传 #

不要在Controller层直接调用上传服务，所有上传请求仅生成任务加入队列后立即返回：

// upload.controller.ts
import { Controller, Post, UploadedFile, UseInterceptors } from '@nestjs/common';
import { FileInterceptor } from '@nestjs/platform-express';
import { InjectQueue } from '@nestjs/bull';
import { Queue } from 'bull';

@Controller('upload')
export class UploadController {
  constructor(@InjectQueue('dropbox-upload') private uploadQueue: Queue) {}

  @Post()
  @UseInterceptors(FileInterceptor('file', { dest: './temp-uploads' }))
  async uploadFile(@UploadedFile() file: Express.Multer.File) {
    // 提交任务到队列，配置失败重试规则
    await this.uploadQueue.add(
      {
        localPath: file.path,
        dropboxPath: `/user-uploads/${file.filename}`
      },
      {
        attempts: 3, // 上传失败最多重试3次
        backoff: {
          type: 'exponential', // 指数退避重试，每次重试间隔翻倍
          delay: 2000 // 首次重试延迟2秒
        }
      }
    );

    // *如果需要接口同步返回上传结果，可取消注释下行，注意设置合理的接口超时时间
    // await job.waitUntilFinished();

    return { code: 0, message: '上传任务已提交，处理中' };
  }
}

可选：强制固定间隔执行任务 #

如果需要每个上传任务之间强制间隔固定时间（比如每2秒执行1个任务），可以在批量添加任务时手动设置延迟：

// 批量提交任务时按顺序设置延迟
const uploadTasks = [/* 你的任务列表 */];
for (let i = 0; i < uploadTasks.length; i++) {
  await this.uploadQueue.add(uploadTasks[i], {
    delay: i * 2000 // 第N个任务比第一个晚执行 N*2 秒
  });
}

效果验证方法 #

• 模拟100个并发请求，观察任务执行日志，确认每秒执行的任务数和你配置的limiter参数一致
• 触发限流时查看队列状态，确认任务会进入等待状态，不会直接返回Dropbox的too many requests错误
• 上传完成后确认本地临时文件已被正常删除

内容的提问来源于stack exchange，提问作者Mateusz Świątek