4. 集成熔断器：Bull + ioredis 与自定义装饰器

熔断器（Circuit Breaker）是微服务架构中防止级联故障的关键模式。当下游服务持续失败时，熔断器会快速拒绝请求，避免调用方线程阻塞和资源耗尽，给下游恢复时间。本章将基于 @nestjs/bull（任务队列）与 ioredis 实现基础熔断，并进一步封装为自定义装饰器，让熔断逻辑与业务代码解耦。

4.1 熔断器核心状态机

熔断器维护三个状态：CLOSED（关闭）、OPEN（打开）、HALF_OPEN（半开）。状态转换由失败计数、阈值和超时时间驱动。

4.2 基于 @nestjs/bull 与 ioredis 的基础熔断实现

我们使用 @nestjs/bull 管理任务队列，ioredis 存储熔断状态（失败计数、状态、时间戳）。Redis 的原子操作和过期机制天然适合熔断器计数与超时。

4.2.1 安装依赖

npm install @nestjs/bull bull ioredis
npm install --save-dev @types/ioredis

4.2.2 创建熔断器服务

import { Injectable } from '@nestjs/common';
import Redis from 'ioredis';

@Injectable()
export class CircuitBreakerService {
  private readonly redis: Redis;
  private readonly failureThreshold = 5;       // 失败阈值
  private readonly timeoutSeconds = 30;        // 熔断超时（秒）
  private readonly halfOpenMaxRequests = 3;    // 半开最大探测数

  constructor() {
    this.redis = new Redis({ host: 'localhost', port: 6379 });
  }

  // 获取当前熔断状态
  async getState(serviceKey: string): Promise<string> {
    const state = await this.redis.get(`cb:${serviceKey}:state`);
    return state || 'CLOSED';
  }

  // 记录失败并判断是否触发熔断
  async recordFailure(serviceKey: string): Promise<void> {
    const current = await this.redis.incr(`cb:${serviceKey}:failures`);
    await this.redis.expire(`cb:${serviceKey}:failures`, this.timeoutSeconds);
    if (current >= this.failureThreshold) {
      await this.redis.set(`cb:${serviceKey}:state`, 'OPEN');
      await this.redis.expire(`cb:${serviceKey}:state`, this.timeoutSeconds);
    }
  }

  // 记录成功（重置计数）
  async recordSuccess(serviceKey: string): Promise<void> {
    await this.redis.del(`cb:${serviceKey}:failures`);
    await this.redis.set(`cb:${serviceKey}:state`, 'CLOSED');
  }

  // 检查是否允许请求通过
  async allowRequest(serviceKey: string): Promise<boolean> {
    const state = await this.getState(serviceKey);
    if (state === 'CLOSED') return true;
    if (state === 'OPEN') {
      // 检查是否超时，超时则进入 HALF_OPEN
      const ttl = await this.redis.ttl(`cb:${serviceKey}:state`);
      if (ttl <= 0) {
        await this.redis.set(`cb:${serviceKey}:state`, 'HALF_OPEN');
        return true;
      }
      return false;
    }
    // HALF_OPEN：限制探测请求数
    const count = await this.redis.incr(`cb:${serviceKey}:halfopen_count`);
    await this.redis.expire(`cb:${serviceKey}:halfopen_count`, 10);
    return count <= this.halfOpenMaxRequests;
  }
}

4.2.3 在 Bull 处理器中集成熔断

import { Processor, Process } from '@nestjs/bull';
import { Job } from 'bull';
import { CircuitBreakerService } from './circuit-breaker.service';

@Processor('payment')
export class PaymentProcessor {
  constructor(private readonly cbService: CircuitBreakerService) {}

  @Process('charge')
  async handleCharge(job: Job) {
    const serviceKey = 'payment:charge';
    const allowed = await this.cbService.allowRequest(serviceKey);
    if (!allowed) {
      throw new Error('CircuitBreaker: request blocked');
    }
    try {
      // 模拟下游调用
      await this.callDownstream(job.data);
      await this.cbService.recordSuccess(serviceKey);
    } catch (err) {
      await this.cbService.recordFailure(serviceKey);
      throw err;
    }
  }

  private async callDownstream(data: any): Promise<void> {
    // 实际 HTTP/RPC 调用
  }
}

4.3 自定义装饰器实现熔断逻辑

为了消除重复的熔断检查代码，我们封装一个 @CircuitBreaker 装饰器，自动在方法执行前后进行状态判断和计数。

4.3.1 装饰器定义

import { SetMetadata } from '@nestjs/common';

export const CIRCUIT_BREAKER_KEY = 'circuit_breaker';

export interface CircuitBreakerOptions {
  serviceKey: string;
  failureThreshold?: number;
  timeoutSeconds?: number;
}

export const CircuitBreaker = (options: CircuitBreakerOptions) =>
  SetMetadata(CIRCUIT_BREAKER_KEY, options);

4.3.2 装饰器执行拦截器

import {
  Injectable,
  ExecutionContext,
  CallHandler,
  NestInterceptor,
  Inject,
} from '@nestjs/common';
import { Reflector } from '@nestjs/core';
import { Observable, from, throwError } from 'rxjs';
import { switchMap, catchError } from 'rxjs/operators';
import { CircuitBreakerService } from './circuit-breaker.service';
import { CIRCUIT_BREAKER_KEY, CircuitBreakerOptions } from './circuit-breaker.decorator';

@Injectable()
export class CircuitBreakerInterceptor implements NestInterceptor {
  constructor(
    private readonly reflector: Reflector,
    private readonly cbService: CircuitBreakerService,
  ) {}

  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    const options = this.reflector.get<CircuitBreakerOptions>(
      CIRCUIT_BREAKER_KEY,
      context.getHandler(),
    );
    if (!options) {
      return next.handle();
    }
    const { serviceKey } = options;

    return from(this.cbService.allowRequest(serviceKey)).pipe(
      switchMap((allowed) => {
        if (!allowed) {
          return throwError(() => new Error(`CircuitBreaker: ${serviceKey} is OPEN`));
        }
        return next.handle().pipe(
          catchError(async (err) => {
            await this.cbService.recordFailure(serviceKey);
            throw err;
          }),
          switchMap(async (result) => {
            await this.cbService.recordSuccess(serviceKey);
            return result;
          }),
        );
      }),
    );
  }
}

4.3.3 在 Controller 中使用

import { Controller, Post, Body, UseInterceptors } from '@nestjs/common';
import { CircuitBreaker } from './circuit-breaker.decorator';
import { CircuitBreakerInterceptor } from './circuit-breaker.interceptor';

@Controller('orders')
@UseInterceptors(CircuitBreakerInterceptor)
export class OrderController {
  @Post()
  @CircuitBreaker({ serviceKey: 'order:create', failureThreshold: 3, timeoutSeconds: 20 })
  async createOrder(@Body() dto: any) {
    // 业务逻辑，熔断由装饰器自动处理
    return { success: true };
  }
}

4.4 方案对比

4.5 避坑指南

4.6 架构总览

下图展示了熔断器在 NestJS 微服务中的完整集成位置：