NestJS-Telegraf开发的Telegram Bot无法处理并发请求问题求助

问题原因 #

问题出在Telegraf默认的更新处理机制：它是串行处理所有 incoming 更新的。哪怕你的onRandom是异步方法，Telegraf的事件循环会等待当前方法内的异步操作完成后，才会处理下一个用户的请求。当asyncRequestToGetData耗时10秒以上时，后续请求会全部排队等待，导致机器人无法并发响应。

解决方案 #

方案1：开启Telegraf多Worker模式

直接在TelegrafModule的配置中添加launchOptions，指定worker数量，让Telegraf用多线程并行处理请求：

TelegrafModule.forRootAsync({
  imports: [ConfigModule],
  useFactory: (configService: ConfigService) => ({
    token: configService.get<string>('TELEGRAM_BOT_TOKEN'),
    // 开启多worker，数量可根据服务器配置调整
    launchOptions: {
      worker: 4, // 示例：启动4个并行worker
      polling: {
        timeout: 30, // 保持polling配置一致性
      },
    },
  }),
  inject: [ConfigService],
}),

这个方法能直接解决串行阻塞问题，适合IO/CPU密集型场景，无需改动业务代码。

方案2：异步解耦耗时操作（生产环境推荐）

通过任务队列将耗时操作与请求处理解耦，先回复用户“正在处理”，待任务完成后再推送结果：

1. 安装依赖：

npm install bullmq @nestjs/bull

2. 在AppModule中注册队列：

import { BullModule } from '@nestjs/bull';

@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true }),
    BullModule.forRoot({
      redis: {
        host: 'localhost',
        port: 6379,
      },
    }),
    BullModule.registerQueue({ name: 'random-task' }),
    TelegrafModule.forRootAsync({ /* 原配置 */ }),
    HttpModule,
  ],
  controllers: [AppController],
  providers: [AppUpdate, RandomTaskProcessor],
})
export class AppModule {}

3. 创建任务处理器：

import { Processor, Process } from '@nestjs/bull';
import { Job } from 'bullmq';
import { Injectable } from '@nestjs/common';
import { Telegraf } from 'telegraf';

@Processor('random-task')
@Injectable()
export class RandomTaskProcessor {
  constructor(private readonly telegraf: Telegraf) {}

  @Process()
  async handleRandomTask(job: Job) {
    const { chatId, text } = job.data;
    try {
      const resp = await asyncRequestToGetData(text);
      await this.telegraf.telegram.sendMessage(chatId, resp);
    } catch (err) {
      await this.telegraf.telegram.sendMessage(chatId, '处理失败，请重试');
    }
  }
}

4. 修改AppUpdate中的命令处理逻辑：

import { InjectQueue } from '@nestjs/bull';
import { Queue } from 'bullmq';

@Update()
@Injectable()
export class AppUpdate {
  constructor(@InjectQueue('random-task') private readonly randomQueue: Queue) {}

  @Command('random')
  async onRandom(@Ctx() ctx: Context) {
    // 先回复用户，避免等待感知
    await ctx.reply('正在获取数据，请稍候...');
    // 将任务加入队列异步处理
    await this.randomQueue.add({
      chatId: ctx.chat.id,
      text: ctx.message.text,
    });
  }
}

该方案能实现完全的请求-处理解耦，支持任务重试、失败告警等生产级特性。

方案3：简易后台异步处理（测试/轻量场景）

如果不想引入队列，可直接让耗时操作在后台执行，不阻塞当前请求：

@Command('random')
async onRandom(@Ctx() ctx: Context) {
  await ctx.reply('正在处理，请稍候');
  // 后台执行耗时操作，不等待完成
  (async () => {
    try {
      const resp = await asyncRequestToGetData(ctx.message.text);
      await ctx.reply(resp);
    } catch (err) {
      await ctx.reply('获取数据失败');
    }
  })();
}

注意：该方式无法保证任务持久化，应用重启时未完成的任务会丢失，仅适合简单场景。

内容的提问来源于stack exchange，提问作者Daniil Andreyevich Baunov