NestJS定时任务CSV处理失败重试的状态外置重构咨询

当前代码把重试状态（attempt_no）耦合在业务方法里，导致重试逻辑和CSV处理逻辑混在一起，下面是几个可行的解耦方案，把重试状态完全抽离出FooService：

方案一：自定义重试装饰器（推荐） #

把重试次数、间隔等配置通过装饰器传入，业务方法只关注核心的CSV解析和DB写入，重试状态由装饰器内部维护。

第一步：实现重试装饰器 #

// src/common/decorators/retry.decorator.ts
import { join } from 'path';
import { renameSync } from 'fs';

export function Retry(maxAttempts: number = 3, retryDelayMs: number = 1000) {
  return function (target: any, methodName: string, descriptor: PropertyDescriptor) {
    const originalMethod = descriptor.value;

    descriptor.value = async function (...args: any[]) {
      let attemptCount = 1;
      let lastError: Error;

      while (attemptCount <= maxAttempts) {
        try {
          const result = await originalMethod.apply(this, args);
          // 业务方法返回文件路径时，统一处理成功后的移动逻辑
          if (result?.sourcePath) {
            const processedPath = join('processed_files', result.sourcePath.split('/').pop());
            renameSync(result.sourcePath, processedPath);
            console.log(`文件已移至processed_files目录: ${processedPath}`);
          }
          return result;
        } catch (error) {
          lastError = error as Error;
          console.log(`第${attemptCount}次处理失败: ${lastError.message}`);
          
          attemptCount++;
          if (attemptCount > maxAttempts) break;
          // 重试前等待指定时间
          await new Promise(resolve => setTimeout(resolve, retryDelayMs));
        }
      }

      // 所有重试失败，处理文件移动
      if (lastError && (lastError as any).sourcePath) {
        const failedPath = join('failed_files', (lastError as any).sourcePath.split('/').pop());
        renameSync((lastError as any).sourcePath, failedPath);
        console.log(`文件已移至failed_files目录: ${failedPath}`);
      }

      throw lastError;
    };

    return descriptor;
  };
}

第二步：改造业务方法 #

去掉attempt_no参数，核心逻辑只负责CSV处理，成功时返回文件路径，失败时抛出带文件路径的错误：

// src/services/foo.service.ts
import { Retry } from '../common/decorators/retry.decorator';
import { join } from 'path';

@Injectable()
export class FooService {
  @Retry(3) // 指定最大重试次数为3
  async foo() {
    const sourcePath = join('csv_file_store', 'foo.csv');

    try {
      // 核心逻辑：读取CSV、逐行解析、写入数据库
      // ...

      return {
        status: { statusCode: 200, statusMessage: '处理成功' },
        sourcePath
      };
    } catch (error) {
      // 抛出错误时携带文件路径，供装饰器处理失败逻辑
      const customError = error as Error;
      (customError as any).sourcePath = sourcePath;
      throw customError;
    }
  }

  // bar、foo2等方法同理，只需添加@Retry装饰器即可
}

第三步：调度层无需修改 #

SchedularService的调用逻辑完全不变，重试逻辑由装饰器自动处理：

@Cron("* * * * * *")
async readSomeFiles(){
  await this.fooService.foo();
  await this.fooService.bar();
}

方案二：独立重试工具类 #

把重试逻辑抽象成通用工具方法，业务方法作为回调传入，重试状态由工具类内部维护，适合需要统一修改重试规则的场景。

实现重试工具类 #

// src/common/utils/retry.util.ts
import { join } from 'path';
import { renameSync } from 'fs';

export class RetryUtil {
  static async execute<T>(
    task: () => Promise<T>,
    config: {
      maxAttempts: number;
      sourcePath: string;
      processedDir: string;
      failedDir: string;
      retryDelayMs?: number;
    }
  ): Promise<T> {
    const { maxAttempts, sourcePath, processedDir, failedDir, retryDelayMs = 1000 } = config;
    let attempt = 1;
    let lastError: Error;

    while (attempt <= maxAttempts) {
      try {
        const result = await task();
        // 成功移动文件
        const targetPath = join(processedDir, sourcePath.split('/').pop()!);
        renameSync(sourcePath, targetPath);
        console.log(`文件已移至${processedDir}目录`);
        return result;
      } catch (error) {
        lastError = error as Error;
        console.log(`第${attempt}次尝试失败: ${lastError.message}`);
        
        attempt++;
        if (attempt > maxAttempts) break;
        await new Promise(resolve => setTimeout(resolve, retryDelayMs));
      }
    }

    // 失败移动文件
    const targetPath = join(failedDir, sourcePath.split('/').pop()!);
    renameSync(sourcePath, targetPath);
    console.log(`文件已移至${failedDir}目录`);
    throw lastError;
  }
}

改造业务方法 #

调用工具类的execute方法，传入核心处理逻辑和配置：

// src/services/foo.service.ts
import { RetryUtil } from '../common/utils/retry.util';
import { join } from 'path';

@Injectable()
export class FooService {
  async foo() {
    const sourcePath = join('csv_file_store', 'foo.csv');

    return RetryUtil.execute(
      async () => {
        // 核心逻辑：读取CSV、解析、写入DB
        // ...
        return { status: { statusCode: 200, statusMessage: '处理成功' } };
      },
      {
        maxAttempts: 3,
        sourcePath,
        processedDir: 'processed_files',
        failedDir: 'failed_files'
      }
    );
  }
}

方案三：NestJS拦截器（全局场景） #

如果需要为多个服务的方法统一添加重试策略，可以用NestJS拦截器结合元数据实现，不过文件移动逻辑需要额外配合异常过滤器处理：

实现重试拦截器和元数据装饰器 #

// src/common/interceptors/retry.interceptor.ts
import { Injectable, NestInterceptor, ExecutionContext, CallHandler } from '@nestjs/common';
import { Observable, throwError } from 'rxjs';
import { retryWhen, delay, scan } from 'rxjs/operators';
import { Reflector } from '@nestjs/core';

@Injectable()
export class RetryInterceptor implements NestInterceptor {
  constructor(private reflector: Reflector) {}

  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    const maxAttempts = this.reflector.get<number>('retry_max_attempts', context.getHandler()) || 3;
    const retryDelayMs = this.reflector.get<number>('retry_delay', context.getHandler()) || 1000;

    return next.handle().pipe(
      retryWhen(errors =>
        errors.pipe(
          scan((attempts) => {
            if (attempts >= maxAttempts) {
              throw new Error(`超过最大重试次数(${maxAttempts})`);
            }
            console.log(`第${attempts + 1}次重试`);
            return attempts + 1;
          }, 0),
          delay(retryDelayMs)
        )
      )
    );
  }
}

// 配套装饰器
export function RetryConfig(maxAttempts: number, retryDelayMs?: number) {
  return function (target: any, methodName: string) {
    Reflect.defineMetadata('retry_max_attempts', maxAttempts, target[methodName]);
    if (retryDelayMs) {
      Reflect.defineMetadata('retry_delay', retryDelayMs, target[methodName]);
    }
  };
}

注册拦截器并标记方法 #

在模块中注册拦截器，然后在业务方法上添加装饰器：

// src/app.module.ts
import { Module } from '@nestjs/common';
import { APP_INTERCEPTOR } from '@nestjs/core';
import { RetryInterceptor } from './common/interceptors/retry.interceptor';

@Module({
  providers: [
    {
      provide: APP_INTERCEPTOR,
      useClass: RetryInterceptor,
    },
    FooService,
    SchedularService
  ],
})
export class AppModule {}

// src/services/foo.service.ts
@Injectable()
export class FooService {
  @RetryConfig(3)
  async foo() {
    const sourcePath = join('csv_file_store', 'foo.csv');
    try {
      // 核心逻辑
      // ...
      const processedPath = join('processed_files', 'foo.csv');
      renameSync(sourcePath, processedPath);
      return { status: { statusCode: 200, statusMessage: '处理成功' } };
    } catch (error) {
      // 结合异常过滤器处理失败后的文件移动
      throw error;
    }
  }
}

选型总结 #

• 装饰器方案：最简洁，侵入性最低，业务代码干净，适合单个或少量方法的重试需求。
• 工具类方案：灵活性高，方便统一修改重试规则，适合大量同类业务方法的场景。
• 拦截器方案：适合全局统一重试策略，需要配合异常过滤器处理文件移动，配置稍复杂。

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