分布式

概述

当前后端架构采用单体应用 + 分布式基础设施的混合模式。核心业务逻辑集中在一个 Express 进程中，但通过 Redis 和 RabbitMQ 等中间件实现跨实例的分布式能力。

这种架构在单实例部署下工作良好，但若要扩展为多实例部署，现有适配存在一些需要改进的地方。

部署架构示意

┌─────────────────────────────────────────────────────────┐
│                       负载均衡                          │
│                   Nginx / 反向代理                       │
└─────────────────────────────────────────────────────────┘
          │                  │                  │
          ▼                  ▼                  ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│  Express 实例 1  │ │  Express 实例 2  │ │  Express 实例 3  │
│  (server:port)   │ │  (server:port)   │ │  (server:port)   │
└──────┬──────────┘ └──────┬──────────┘ └──────┬──────────┘
       │                   │                   │
       └──────────┬────────┴────────┬──────────┘
                  │                 │
                  ▼                 ▼
          ┌─────────────┐  ┌──────────────┐
          │    Redis    │  │   RabbitMQ   │
          │  (会话/缓存  │  │  (异步消息)  │
          │   限流/锁)   │  │              │
          └─────────────┘  └──────────────┘
                  │
                  ▼
          ┌─────────────┐
          │   MySQL     │
          │  (主库)     │
          └─────────────┘

已实现的分布式适配

Redis 分布式缓存

Redis 作为核心分布式基础设施，承担缓存、限流、会话管理等多重角色。通过 RedisUtils.mts 封装了统一的客户端连接管理和操作接口。

Redis 工具类封装（@src/utils/RedisUtils.mts）

• 单例模式：getRedisClient() 确保全局只有一个 Redis 客户端实例
• 连接管理：connectRedis() 在应用启动时初始化连接
• 基础 KV 操作：redisSet、redisGet、redisDel、redisExists
• Sorted Set 操作：redisZAdd、redisZRem、redisZMembers 等，用于有序集合场景

import { createClient, type RedisClientType } from "redis";
import { REDIS_URL, REDIS_PASSWORD } from "#utils/ConstantUtils";
import { logger } from "#utils/LogUtils";

let redisClient: RedisClientType;

export const getRedisClient = (): RedisClientType => {
  if (!redisClient) {
    redisClient = createClient({
      url: REDIS_URL,
      password: REDIS_PASSWORD || undefined,
      pingInterval: 30000,
      socket: {
        connectTimeout: 5000,
        reconnectStrategy: (retries: number) => {
          const jitter = Math.floor(Math.random() * 200);
          const delay = Math.min(Math.pow(2, retries) * 50, 2000);
          return delay + jitter;
        },
      },
    });

    redisClient.on("error", (err) => {
      logger.error("Redis Client Error", err);
    });

    redisClient.on("connect", () => {
      logger.info("Redis Client Connected");
    });

    redisClient.on("ready", () => {
      logger.info("Redis Client Ready");
    });

    redisClient.on("reconnecting", () => {
      logger.warn("Redis Client Reconnecting...");
    });
  }
  return redisClient;
};

export const connectRedis = async (): Promise<RedisClientType> => {
  const client = getRedisClient();
  if (!client.isOpen) {
    await client.connect();
  }
  return client;
};

export const redisSet = async (
  key: string,
  value: string | Buffer,
  expireSeconds?: number,
): Promise<void> => {
  const client = getRedisClient();
  if (expireSeconds) {
    await client.set(key, value, { EX: expireSeconds });
  } else {
    await client.set(key, value);
  }
};

export const redisGet = async (key: string): Promise<string | null> => {
  const client = getRedisClient();
  return client.get(key);
};

export const redisDel = async (key: string): Promise<void> => {
  const client = getRedisClient();
  await client.del(key);
};

export const redisExists = async (key: string): Promise<boolean> => {
  const client = getRedisClient();
  const result = await client.exists(key);
  return result === 1;
};

// #region Sorted Set 操作
export const redisZAdd = async (key: string, score: number, member: string): Promise<void> => {
  const client = getRedisClient();
  await client.zAdd(key, { score, value: member });
};

export const redisZRem = async (key: string, member: string): Promise<void> => {
  const client = getRedisClient();
  await client.zRem(key, member);
};

export const redisZIsMember = async (key: string, member: string): Promise<boolean> => {
  const client = getRedisClient();
  const score = await client.zScore(key, member);
  return score !== null;
};

export const redisZCount = async (key: string): Promise<number> => {
  const client = getRedisClient();
  return client.zCount(key, 0, Date.now());
};

export const redisZRemRangeByRank = async (key: string, start: number, stop: number): Promise<void> => {
  const client = getRedisClient();
  await client.zRemRangeByRank(key, start, stop);
};

export const redisZMembers = async (key: string): Promise<string[]> => {
  const client = getRedisClient();
  return client.zRange(key, 0, -1);
};

// #region trackUserLogin
const TRACK_USER_LOGIN_SCRIPT = `
local members = redis.call("ZRANGE", KEYS[1], 0, -1)
for _, existingMember in ipairs(members) do
  if string.sub(existingMember, 1, 6) ~= "login:" then
    redis.call("ZREM", KEYS[1], existingMember)
  end
end
redis.call("ZADD", KEYS[1], ARGV[1], ARGV[2])
redis.call("ZREMRANGEBYRANK", KEYS[1], 0, -(tonumber(ARGV[3]) + 1))
redis.call("EXPIRE", KEYS[1], ARGV[4])
return 1
`;

export const redisTrackUserLogin = async ({
  key,
  score,
  member,
  maxSessions,
  expireSeconds,
}: {
  key: string;
  score: number;
  member: string;
  maxSessions: number;
  expireSeconds: number;
}): Promise<void> => {
  const client = getRedisClient();
  await client.eval(TRACK_USER_LOGIN_SCRIPT, {
    keys: [key],
    arguments: [String(score), member, String(maxSessions), String(expireSeconds)],
  });
};
// #endregion trackUserLogin
// #endregion Sorted Set 操作

// #region tokenBucketRateLimit
const TOKEN_BUCKET_SCRIPT = `
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

local bucket = redis.call("HMGET", key, "tokens", "last_refill")
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now

local elapsed = now - last_refill
if elapsed > 0 then
    local refill = math.floor(elapsed * rate)
    if refill > 0 then
        tokens = math.min(capacity, tokens + refill)
        last_refill = now
    end
end

if tokens >= 1 then
    tokens = tokens - 1
    redis.call("HMSET", key, "tokens", tokens, "last_refill", last_refill)
    redis.call("EXPIRE", key, math.ceil(capacity / rate) + 1)
    return 1
else
    redis.call("HMSET", key, "tokens", tokens, "last_refill", last_refill)
    redis.call("EXPIRE", key, math.ceil(capacity / rate) + 1)
    return 0
end
`;

/**
 * 检查令牌桶是否允许通过
 * @param bucketKey Redis key
 * @param capacity 桶容量（最大突发）
 * @param refillRate 每秒补充速率（QPS）
 * @returns true=允许，false=限流
 */
export const checkTokenBucket = async (
  bucketKey: string,
  capacity: number,
  refillRate: number,
): Promise<boolean> => {
  const client = getRedisClient();
  const now = Math.floor(Date.now() / 1000);
  const result = await client.eval(TOKEN_BUCKET_SCRIPT, {
    keys: [bucketKey],
    arguments: [String(capacity), String(refillRate), String(now)],
  });
  return result === 1;
};
// #endregion tokenBucketRateLimit

用户登录会话追踪

通过 Lua 脚本在 Redis 中维护用户登录会话的 Sorted Set，实现每个用户的最大会话数控制（maxSessions），自动清理过期会话。

用户登录会话追踪 Lua 脚本（@src/utils/RedisUtils.mts）

-- 清理非 login: 前缀的脏数据
-- 添加新会话
-- 按排名移除超出 maxSessions 的旧会话
-- 设置 key 过期时间

const TRACK_USER_LOGIN_SCRIPT = `
local members = redis.call("ZRANGE", KEYS[1], 0, -1)
for _, existingMember in ipairs(members) do
  if string.sub(existingMember, 1, 6) ~= "login:" then
    redis.call("ZREM", KEYS[1], existingMember)
  end
end
redis.call("ZADD", KEYS[1], ARGV[1], ARGV[2])
redis.call("ZREMRANGEBYRANK", KEYS[1], 0, -(tonumber(ARGV[3]) + 1))
redis.call("EXPIRE", KEYS[1], ARGV[4])
return 1
`;

export const redisTrackUserLogin = async ({
  key,
  score,
  member,
  maxSessions,
  expireSeconds,
}: {
  key: string;
  score: number;
  member: string;
  maxSessions: number;
  expireSeconds: number;
}): Promise<void> => {
  const client = getRedisClient();
  await client.eval(TRACK_USER_LOGIN_SCRIPT, {
    keys: [key],
    arguments: [String(score), member, String(maxSessions), String(expireSeconds)],
  });
};

分布式限流

实现了两种分布式限流策略，均基于 Redis 实现跨实例共享计数器。

滑动窗口限流

基于 express-rate-limit + rate-limit-redis，使用 Redis 存储计数器，多实例共享同一限流窗口。

滑动窗口限流中间件（@src/middlewares/RateLimitMiddleware.mts）

import { rateLimit } from "express-rate-limit";
import RedisStore from "rate-limit-redis";
import { createApiResponse } from "#utils/ResponseUtils";
import { connectRedis } from "#utils/RedisUtils";
import { wrapAsyncMiddleware } from "#utils/MiddlewareUtils";

/**
 * express限速中间件
 * 不仅限/api开头的接口请求
 * 也限制html页面请求（防爬虫）
 * 使用 Redis 存储实现分布式限流，多实例共享计数器
 */
export function rateLimitMiddleware(limitPer15Minutes: number) {
  return wrapAsyncMiddleware(rateLimit({
    windowMs: 15 * 60 * 1000, // 15 minutes
    limit: limitPer15Minutes, // Limit each IP to 100 requests per `window` (here, per 15 minutes).
    standardHeaders: "draft-8", // draft-6: `RateLimit-*` headers; draft-7 & draft-8: combined `RateLimit` header
    legacyHeaders: false, // Disable the `X-RateLimit-*` headers.
    statusCode: 429,
    message: createApiResponse({
      code: 429,
      data: null,
      message: "请求太频繁了，请稍后再试",
    }),
    store: new RedisStore({
      sendCommand: async (...args: string[]) => {
        const client = await connectRedis();
        return client.sendCommand(args);
      },
    }),
  }));
}

在 app.mts 中注册，全局生效，15 分钟窗口内限制请求数。

令牌桶限流

通过 Lua 脚本实现原子化的令牌桶算法，支持可配置的桶容量（Burst）和补充速率（QPS），以及分布式告警冷却机制。

令牌桶限流中间件（@src/middlewares/TokenBucketRateLimitMiddleware.mts）

• createTokenBucketRateLimiter 高阶函数，可配置容量、速率、key 提取函数
• 底层通过 RedisUtils.mts 中的 checkTokenBucket 函数执行 Lua 脚本
• 超限时使用 SET NX 实现分布式告警冷却，同一 key 在冷却期内不重复告警

import type { Request, Response, NextFunction } from "express";
import { checkTokenBucket, getRedisClient } from "#utils/RedisUtils";
import { sendSystemEmailSafe } from "#utils/EmailUtils";
import { wrapAsyncMiddleware } from "#utils/MiddlewareUtils";

type KeyResolver = (req: Request) => string;

interface TokenBucketConfig {
  /** 桶容量（最大突发） */
  capacity: number;
  /** 每秒补充速率（QPS） */
  refillRate: number;
  /** 从请求中提取限流 key 的函数 */
  keyResolver: KeyResolver;
  /** Redis key 前缀，默认 "token_bucket" */
  cacheKeyPrefix?: string;
  /** 告警冷却时间（秒），默认 300（5 分钟） */
  alertCooldownSeconds?: number;
  /** 告警邮件主题前缀，默认 "令牌桶限流告警" */
  alertSubject?: string;
}

/**
 * 创建令牌桶限流中间件（高阶函数）
 * @param config 令牌桶配置
 * @returns Express 中间件
 */
export function createTokenBucketRateLimiter(config: TokenBucketConfig) {
  const {
    capacity,
    refillRate,
    keyResolver,
    cacheKeyPrefix = "token_bucket",
    alertCooldownSeconds = 300,
    alertSubject = "令牌桶限流告警",
  } = config;

  return wrapAsyncMiddleware(async (req: Request, res: Response, next: NextFunction) => {
    const key = keyResolver(req);
    const bucketKey = `${cacheKeyPrefix}:${key}`;
    const allowed = await checkTokenBucket(bucketKey, capacity, refillRate);

    if (allowed) {
      next();
      return;
    }

    // 超限：丢弃请求，邮件告警（使用 Redis SET NX 实现分布式冷却，同一 key 在冷却期内不重复告警）
    const alertKey = `${cacheKeyPrefix}:alert_cooldown:${key}`;
    const client = getRedisClient();
    const alertSent = await client.set(alertKey, "1", {
      EX: alertCooldownSeconds,
      NX: true,
    });
    if (alertSent) {
      sendSystemEmailSafe({
        subject: alertSubject,
        text: `key=${key} 触发令牌桶限流，请求已被丢弃。当前配置：QPS=${refillRate}, Burst=${capacity}`,
        req,
      });
    }

    res.status(429).json({ code: 429, message: "too many requests" });
  });
}

令牌桶 Lua 脚本（@src/utils/RedisUtils.mts）：

const TOKEN_BUCKET_SCRIPT = `
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

local bucket = redis.call("HMGET", key, "tokens", "last_refill")
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now

local elapsed = now - last_refill
if elapsed > 0 then
    local refill = math.floor(elapsed * rate)
    if refill > 0 then
        tokens = math.min(capacity, tokens + refill)
        last_refill = now
    end
end

if tokens >= 1 then
    tokens = tokens - 1
    redis.call("HMSET", key, "tokens", tokens, "last_refill", last_refill)
    redis.call("EXPIRE", key, math.ceil(capacity / rate) + 1)
    return 1
else
    redis.call("HMSET", key, "tokens", tokens, "last_refill", last_refill)
    redis.call("EXPIRE", key, math.ceil(capacity / rate) + 1)
    return 0
end
`;

/**
 * 检查令牌桶是否允许通过
 * @param bucketKey Redis key
 * @param capacity 桶容量（最大突发）
 * @param refillRate 每秒补充速率（QPS）
 * @returns true=允许，false=限流
 */
export const checkTokenBucket = async (
  bucketKey: string,
  capacity: number,
  refillRate: number,
): Promise<boolean> => {
  const client = getRedisClient();
  const now = Math.floor(Date.now() / 1000);
  const result = await client.eval(TOKEN_BUCKET_SCRIPT, {
    keys: [bucketKey],
    arguments: [String(capacity), String(refillRate), String(now)],
  });
  return result === 1;
};

Redis 分布式会话

使用 express-session + connect-redis（RedisStore）将会话数据存储在 Redis 中，多实例共享会话状态，前缀为 session:。

会话中间件（@src/middlewares/SessionMiddleware.mts）

import session from "express-session";
import { RedisStore } from "connect-redis";
import { JWT_SECRET_KEY, NODE_ENV, SESSION_EXPIRE_MINUTES } from "#utils/ConstantUtils";
import { getRedisClient } from "#utils/RedisUtils";
import { wrapAsyncMiddleware } from "#utils/MiddlewareUtils";

export const SESSION_COOKIE_NAME = "session_id";

export const SESSION_COOKIE_OPTIONS = {
  path: "/",
  httpOnly: true,
  /**
   * 生产环境仅通过 HTTPS 传输
   * 正式环境 nginx 上记得配置：proxy_set_header X-Forwarded-Proto $scheme; # 向后端传递原始请求协议
   */
  secure: NODE_ENV === "production",
  domain: NODE_ENV === "production" ? ".verysites.com" : undefined,
  // 限制跨站请求，防止 CSRF 攻击
  sameSite: "lax" as const,
};

export const sessionMiddleware = wrapAsyncMiddleware(session({
  secret: JWT_SECRET_KEY || "default-secret-key",
  resave: false,
  saveUninitialized: false,
  name: SESSION_COOKIE_NAME,
  store: new RedisStore({
    client: getRedisClient(),
    prefix: "session:",
  }),
  cookie: {
    ...SESSION_COOKIE_OPTIONS,
    maxAge: SESSION_EXPIRE_MINUTES * 60 * 1000, // 会话有效期，单位毫秒
  },
}));

RabbitMQ 消息队列

RabbitMQ 作为异步消息队列主干，承载 10 个队列，主要用于日志批量异步写入和耗时任务异步处理。

RabbitMQ 工具类封装（@src/utils/RabbitMQUtils.mts）

• 连接管理：连接失败后自动重试（5 秒间隔）
• 队列声明缓存：declaredQueues Set 缓存已声明队列，避免重复声明
• 单条发送：sendDataToRabbitMQQueue，支持持久化消息
• 批量发送：sendBatchToRabbitMQQueue，支持并行/串行两种模式
• 消息消费：consumeRabbitMQQueue，支持手动 ACK、prefetch 限流、批量 flush

import amqp, { type ConsumeMessage, type ChannelModel, type Channel, type Options } from "amqplib";
import { RABBITMQ_URL } from "#utils/ConstantUtils";
import { logger } from "#utils/LogUtils";
import { sleep } from "#utils/TimeUtils";

let channel: Channel | null = null;
const declaredQueues = new Set<string>();

//  连接 RabbitMQ
async function connectRabbitMQ(): Promise<void> {
  try {
    logger.info("RabbitMQ 正在连接...");
    const connection: ChannelModel = await amqp.connect(RABBITMQ_URL);
    channel = await connection.createChannel();
    for (const queue of declaredQueues) {
      // 持久化队列
      await channel.assertQueue(queue, { durable: true });
    }
    logger.info("RabbitMQ 已连接");
  } catch (error) {
    const timeWait = 5000;
    logger.error(`RabbitMQ 连接失败，将于${timeWait}ms后尝试重连`, error);
    await sleep(timeWait);
    await connectRabbitMQ();
  }
}

/**
 * 确保队列已声明
 * @param channel RabbitMQ 通道对象
 * @param queueName 队列名称
 * @param options 队列配置选项
 */
async function ensureQueue(channel: Channel, queueName: string, options: Options.AssertQueue) {
  if (declaredQueues.has(queueName)) {
    return;
  }

  await channel.assertQueue(queueName, options);
  declaredQueues.add(queueName);
}

// 发送日志等消息
export async function sendDataToRabbitMQQueue<T extends object>(
  queueName: string,
  dataToStringify: T,
) {
  try {
    logger.info(`尝试发送消息至队列：${queueName}`);
    while (!channel) {
      await connectRabbitMQ();
    }
    await ensureQueue(channel, queueName, { durable: true });
    const msg = Buffer.from(JSON.stringify(dataToStringify));
    const success = channel.sendToQueue(queueName, msg, {
      // 持久化消息
      persistent: true,
    });
    if (!success) {
      logger.error(`[success=${success}]: RabbitMQ Producer 未能成功发送消息至队列 ${queueName}`);
      return;
    }
    logger.info(`[success=${success}]: RabbitMQ Producer 已成功发送消息至队列 ${queueName}`);
  } catch (error) {
    logger.error(`RabbitMQ Producer 发送失败，目标队列名为 ${queueName}`, error);
  }
}

/**
 * 批量发送消息到 RabbitMQ 队列
 * 相比逐条发送，减少网络开销和连接占用
 * @param queueName 队列名称
 * @param dataList 数据列表
 * @param parallel 是否并行发送，默认 false（串行发送更稳定）
 */
export async function sendBatchToRabbitMQQueue<T extends object>(
  queueName: string,
  dataList: T[],
  parallel = false,
): Promise<{ success: number; failed: number }> {
  const result = { success: 0, failed: 0 };

  if (!dataList || dataList.length === 0) {
    return result;
  }

  try {
    while (!channel) {
      await connectRabbitMQ();
    }
    await ensureQueue(channel, queueName, { durable: true });

    if (parallel) {
      // 并行发送：使用 Promise.allSettled 确保所有消息都尝试发送
      const results = dataList.map((data) => {
        const msg = Buffer.from(JSON.stringify(data));
        return channel!.sendToQueue(queueName, msg, { persistent: true });
      });
      results.forEach((isSent) => {
        if (isSent) {
          result.success++;
        } else {
          result.failed++;
        }
      });
    } else {
      // 串行发送：更稳定，适合高吞吐场景
      for (const data of dataList) {
        try {
          const msg = Buffer.from(JSON.stringify(data));
          const success = channel.sendToQueue(queueName, msg, { persistent: true });
          if (success) {
            result.success++;
          } else {
            result.failed++;
          }
        } catch {
          result.failed++;
        }
      }
    }

    logger.info(
      `RabbitMQ 批量发送完成，队列: ${queueName}，成功: ${result.success}，失败: ${result.failed}`,
    );
  } catch (error) {
    logger.error(`RabbitMQ 批量发送失败，目标队列名为 ${queueName}`, error);
    result.failed = dataList.length - result.success;
  }

  return result;
}

// 消费指定队列的消息
export async function consumeRabbitMQQueue<T extends object>(
  payload: ParamsConsumeRabbitMQQueue<T>,
) {
  const { needAck, queueName, channelPrefetchSize, flushInterval, flushBatchSize, flushCallback } =
    payload;
  logger.info(`开启队列消费脚本用于消费队列：${queueName}`);
  while (!channel) {
    await connectRabbitMQ();
  }
  await ensureQueue(channel, queueName, { durable: true });

  // 通过 channel.prefetch(n)，你可以告诉 RabbitMQ：“我一次最多只能处理 n 条消息，请等我确认（ack）后再发新的。”
  await channel.prefetch(channelPrefetchSize);

  const logBuffer: T[] = [];

  async function flushBuffer() {
    if (logBuffer.length === 0) {
      return;
    }
    const logs = [...logBuffer];
    logBuffer.length = 0;

    try {
      await flushCallback(logs);
      logger.info(`✅ 批量写入 ${logs.length} 条日志`);
    } catch (error) {
      logger.error(`❌ RabbitMQ 批量消费队列${queueName}失败`, error, logs);
      // 失败后重放回缓冲区（或记录到文件）
      // logBuffer.push(...logs);
    }
  }

  setInterval(flushBuffer, flushInterval);

  const consumeConfig: Options.Consume = {
    /**
     * false: 不自动确认消息已处理，改为代码中手动确认，避免消息直接在被发送后从消息队列中删除掉（即便未被正常消费）
     * true: 自动确认消息已处理，提高效率
     */
    noAck: !needAck,
  };
  await channel.consume(
    queueName,
    async (msg: ConsumeMessage | null) => {
      if (!msg) {
        return;
      }
      const data = JSON.parse(msg.content.toString()) as T;
      logBuffer.push(data);

      // 手动确认消息，确保不丢失
      while (!channel) {
        await connectRabbitMQ();
      }
      if (needAck) {
        channel.ack(msg);
      }

      // 达到批量大小，立即写入
      if (logBuffer.length >= flushBatchSize) {
        await flushBuffer();
      }
    },
    consumeConfig,
  );
}

队列一览

队列名	用途	消费者	消费配置
b_action	Bugs 用户行为日志	consumers/bugs.mts	needAck=false, prefetch=1000, flush 1s/1000条
b_api	Bugs API 调用日志	同上	同上
b_bug	Bugs 异常日志	同上	同上
b_bv	Bugs 浏览器版本	同上	同上
b_pv	Bugs PV 日志	同上	同上
b_uv	Bugs UV 日志	同上	同上
b_event	Bugs 事件日志	同上	同上
b_referer	Bugs 来源日志	同上	同上
pdf_toolbox	PDF 处理任务	consumers/pdfToolbox.mts	needAck=false, prefetch=5, flush 3s/10条
paper_summarizer	论文摘要 AI 生成	consumers/paperSummarizer.mts	needAck=false, prefetch=5, flush 3s/10条

Bugs 消费者启动（@consumers/bugs.mts）

8 个队列的消费者并行启动，仅在生产环境生效。

import { NODE_ENV } from "#utils/ConstantUtils";
import { consumeBugQueue } from "#services/bugs/bug";
import { consumeActionQueue } from "#services/bugs/action";
import { consumeApiQueue } from "#services/bugs/api";
import { consumeBvQueue } from "#services/bugs/bv";
import { consumeEventQueue } from "#services/bugs/event";
import { consumePvQueue } from "#services/bugs/pv";
import { consumeUvQueue } from "#services/bugs/uv";
import { consumeRefererQueue } from "#services/bugs/referer";
import { logger } from "#utils/LogUtils";

try {
  if (NODE_ENV === "production") {
    await Promise.all([
      consumeBugQueue(),
      consumeActionQueue(),
      consumeApiQueue(),
      consumeBvQueue(),
      consumePvQueue(),
      consumeUvQueue(),
      consumeEventQueue(),
      consumeRefererQueue(),
    ]);
  }
  logger.info(`[Success]成功启动bugs相关队列消费脚本`);
} catch (error) {
  logger.error(`[Fail]启动bugs相关队列消费脚本失败`, error);
}

PDF 工具箱消费者（@consumers/pdfToolbox.mts）

PDF 处理是重量级任务，prefetch 设为 5 限制并发，flushCallback 中逐条处理。

export async function consumePdfToolboxQueue(): Promise<void> {
  await consumeRabbitMQQueue<PdfToolboxTaskMessage>({
    needAck: false,
    queueName: MQ_QUEUE_NAME_PDF_TOOLBOX,
    channelPrefetchSize: 5,
    flushInterval: 3000,
    flushBatchSize: 10,
    flushCallback: async function (messages) {
      for (const message of messages) {
        await processTask(message);
      }
    },
  });
}

MySQL 事务与行级锁并发控制

在资金敏感操作中使用 knex.transaction + SELECT ... FOR UPDATE 行级锁确保并发安全和数据一致性。

余额系统

balance.mts 中 12 处使用 forUpdate，覆盖充值、消费、退款、冻结、提现等核心资金操作。

消费扣款 — 事务 + 行级锁（@src/services/balance.mts）

await knex.transaction(async (trx) => {
  // 查询并锁定账户，防止超扣
  const account = await trx("balance_account")
    .where({ user_id: userId })
    .forUpdate()
    .first();
  // 检查可用余额（总余额 - 冻结余额）
  // 扣减余额并创建交易记录
});

export async function consumeBalance({
  userId,
  amount,
  businessType,
  businessId,
  remark,
}: {
  userId: number;
  amount: number;
  businessType?: BalanceBusinessType;
  businessId?: number;
  remark?: string;
}): Promise<TReturn<{ transactionId: number; balanceAfter: number }>> {
  try {
    if (amount <= 0) {
      return [new Error("消费金额必须大于0"), undefined];
    }

    let transactionId: number = 0;
    let balanceAfter: number = 0;

    await knex.transaction(async (trx) => {
      // 查询并锁定账户
      const account = await trx("balance_account").where({ user_id: userId }).forUpdate().first();

      if (!account) {
        throw new Error("账户不存在");
      }

      const balanceBefore = Number(account.balance) || 0;
      const frozenBalance = Number(account.frozen_balance) || 0;
      const availableBalance = balanceBefore - frozenBalance; // 可用余额 = 总余额 - 冻结余额

      // 检查可用余额是否足够
      if (availableBalance < amount) {
        throw new Error(
          `余额不足，当前余额: ${balanceBefore}元，冻结余额: ${frozenBalance}元，可用余额: ${availableBalance}元`,
        );
      }

      balanceAfter = balanceBefore - amount;

      // 创建交易记录
      const [id] = await trx("balance_transaction").insert({
        user_id: userId,
        transaction_type: "consume",
        amount: -amount, // 负数表示减少余额
        balance_before: balanceBefore,
        balance_after: balanceAfter,
        status: "success",
        business_type: businessType || null,
        business_id: businessId || null,
        remark: remark || "消费扣款",
      });

      transactionId = id;

      // 更新余额
      await trx("balance_account").where({ user_id: userId }).update({
        balance: balanceAfter,
        last_updated: trx.fn.now(),
      });
    });

    if (transactionId === 0 || balanceAfter === 0) {
      return [new Error("消费扣款失败"), undefined];
    }

    return [null, { transactionId, balanceAfter }];
  } catch (err) {
    logger.error("error in consumeBalance", err);
    return [getError(err), undefined];
  }
}

充值成功处理（@src/services/balance.mts）

await knex.transaction(async (trx) => {
  // 锁定订单，防止重复处理
  const order = await trx("order")
    .where({ order_no: orderNo })
    .forUpdate()
    .first();
  // 锁定余额账户，原子性增加余额
  const account = await trx("balance_account")
    .where({ user_id: userId })
    .forUpdate()
    .first();
  // 创建交易记录 + 更新余额
});

export async function handleRechargeSuccess({
  orderNo,
  ltzfOrderNo,
  tradeOrderNo,
  paySuccessTime,
}: {
  orderNo: string; // 商户订单号（我们的order_no）
  ltzfOrderNo?: string; // 蓝兔支付系统订单号（回调中的order_no）
  tradeOrderNo: string; // 微信支付订单号（回调中的pay_no）
  paySuccessTime: string;
}): Promise<TReturn<void>> {
  try {
    await knex.transaction(async (trx) => {
      // 查询订单
      const order = await trx("order").where({ order_no: orderNo }).forUpdate().first();

      if (!order) {
        throw new Error(`订单不存在: ${orderNo}`);
      }

      // 如果已经处理过，直接返回
      if (order.pay_status === 1) {
        logger.info(`订单 ${orderNo} 已经处理过，跳过`);
        return;
      }

      const userId = order.user_id;
      const rechargeAmount = Number(order.total_fee);

      // 查询或创建余额账户
      const account = await trx("balance_account").where({ user_id: userId }).forUpdate().first();

      const balanceBefore = Number(account?.balance || 0);
      const balanceAfter = Number(BigNumber.sum(balanceBefore, rechargeAmount).toString());

      // 创建余额交易记录
      await trx("balance_transaction").insert({
        user_id: userId,
        transaction_type: "recharge",
        amount: rechargeAmount,
        balance_before: balanceBefore,
        balance_after: balanceAfter,
        status: "success",
        order_id: order.id,
        remark: "充值到余额",
      });

      // 更新或创建余额账户
      if (account) {
        await trx("balance_account").where({ user_id: userId }).update({
          balance: balanceAfter,
          last_updated: trx.fn.now(),
        });
      } else {
        await trx("balance_account").insert({
          user_id: userId,
          balance: balanceAfter,
          frozen_balance: 0,
        });
      }

      // 更新订单状态
      await trx("order")
        .where({ id: order.id })
        .update({
          pay_status: 1, // 支付成功
          task_status: 1, // 任务处理成功
          trade_order_no: tradeOrderNo, // 微信支付订单号
          external_order_no: ltzfOrderNo || "", // 蓝兔支付系统订单号
          pay_success_time: paySuccessTime,
        });

      logger.info(`充值成功: 用户${userId}, 金额${rechargeAmount}元, 订单${orderNo}`);
    });

    return [null, undefined];
  } catch (err) {
    logger.error("error in handleRechargeSuccess", err);
    return [getError(err), undefined];
  }
}

HTTP 代理

在 app.mts 中使用 http-proxy-middleware 实现请求代理转发，支持开发环境前端子应用代理和生产环境外部服务代理。

HTTP 代理配置（@src/app.mts）

// 开发环境：代理前端子应用
// 生产环境：代理 commonrail 服务到外部 URL
const proxyRoute = (routeName, target) => {
  app.use(routeName, createProxyMiddleware({ target, changeOrigin: true, ... }));
};

const proxyRoute = (routeName: string, target: string) => {
  app.use(
    routeName,
    createProxyMiddleware({
      target,
      changeOrigin: true,
      on: {
        proxyReq: (proxyReq, req: ExpressRequest) => {
          const body = req.body;
          if (body) {
            const contentType = req.get("Content-Type") || "";
            if (!contentType.includes("multipart/form-data")) {
              const bodyData = JSON.stringify(body);
              proxyReq.setHeader("Content-Length", Buffer.byteLength(bodyData));
              // stream the content
              proxyReq.write(bodyData);
            }
          }
        },
      },
    }),
  );
};

proxyRoute("/test/commonrail", `https://test.51gonggui.com/commonrail`);
proxyRoute("/production/commonrail", `https://wx.51gonggui.com/commonrail`);

同时配置了 app.set("trust proxy", 1)，支持 Nginx 反向代理场景下获取真实客户端 IP。

WebSocket 实时通信

通过 ws 库实现 WebSocket 服务器，支持消息广播和心跳检测。

WebSocket 路由注册（@src/websockets/index.mts）

import type { WebSocketServer } from "ws";

import wssRoot from "#websockets/root";
import wssMqttMockTunerPad from "#websockets/mqttTunerPad";

export interface WebSocketRoute {
  path: string;
  server: WebSocketServer;
}

export const websocketRoutes: WebSocketRoute[] = [
  { path: "/", server: wssRoot },
  { path: "/mqtt/mock/tunerpad", server: wssMqttMockTunerPad },
];

WebSocket 通用处理（@src/websockets/root.mts）

支持消息收发、广播、心跳检测。

import WebSocket, { WebSocketServer } from "ws";
import { logger } from "#utils/LogUtils";

const wss = new WebSocketServer({ noServer: true });

wss.on("connection", (ws) => {
  console.log("一个客户端成功建立 WebSocket 连接 🤝");

  // 监听客户端发送的消息
  ws.on("message", (data) => {
    const message = Buffer.isBuffer(data)
      ? data.toString("utf-8")
      : typeof data === "string"
        ? data
        : Array.isArray(data)
          ? Buffer.concat(data).toString("utf-8")
          : Buffer.from(data).toString("utf-8");
    logger.info("收到客户端消息：", message);
    // 向客户端回复消息
    ws.send(`服务端已收到：${message}，开始每隔2秒发一次消息 🚀`);

    let count = 0;
    const timer = setInterval(() => {
      count++;
      ws.send(
        `eyJhY3Rpb24iOiJzdHJlYW1fY29sbGVjdF9pdGVtIiwiY29kZSI6MjAwLCJkYXRhIjp7Im5hbWUiOiLnlLXmsaDnlLXljosiLCJ1bml0IjoibVYiLCJ2YWx1ZSI6IjIzNzYwLjAwIn19\n`,
      );
      if (count > 30) {
        clearInterval(timer);
      }
    }, 200);

    // 【可选】广播：向所有已连接的客户端发送消息（除了发送者）
    wss.clients.forEach((client) => {
      if (client.readyState === WebSocket.OPEN && client !== ws) {
        client.send(`[广播] 其他客户端说：${message}`);
      }
    });
  });

  // 监听客户端断开连接
  ws.on("close", () => {
    console.log("客户端断开 WebSocket 连接 👋");
  });

  // 监听 WebSocket 错误
  ws.on("error", (err) => {
    console.error("WebSocket 错误：", err);
  });
});

export default wss;

定时任务调度

使用 node-schedule 实现 9 个定时任务，在 app.mts 启动时注册。

定时任务调度入口（@schedules/index.mts）

仅在生产环境执行，覆盖统计刷新、表优化、记录清理、PPT 任务、预订提醒、告警检测、SSL 证书、文章爬虫、站点地图生成。

import schedule from "node-schedule";
import { getError } from "nsuite";

import { jobOptimizeBugsTable } from "#schedules/jobOptimizeBugsTables";
import { jobBugsStatistics } from "#schedules/jobBugsStatistics";
import { sendSystemEmailSafe } from "#utils/EmailUtils";
import { logger } from "#utils/LogUtils";
import { jobBugsLimitRecords } from "#schedules/jobBugsLimitRecords";
import { jobPPTTaskUpdateStatus } from "#schedules/jobPPTTask";
import { jobBookingReminder } from "#schedules/jobBookingReminder";
import { jobBugsAlert } from "#schedules/jobBugsAlert";
import { jobSSLCertificate } from "#schedules/jobSSLCertificate";
import { runArticleCrawler } from "#schedules/article-crawler";
import { jobGenerateSitemap } from "#schedules/jobSitemap";
import { MODE } from "#utils/ConstantUtils";

export const doScheduleJobs = async () => {
  if (MODE !== "production") {
    logger.info("[doScheduleJobs]", "开发环境，不执行定时任务");
    return;
  }

  await schedule.gracefulShutdown();

  // 每10分钟，刷新一次统计数据
  const intervalJobBugsStatistics = 10;
  for (let i = 0; i < 60; i += intervalJobBugsStatistics) {
    schedule.scheduleJob({ minute: i }, () => {
      // 只需要0点的时候邮件提醒一次就可，不要每10分钟就提醒一次
      const shouldNotify = i === 0 && new Date().getHours() === 0;
      jobBugsStatistics({
        shouldNotify,
        intervalInMinutes: intervalJobBugsStatistics,
      }).catch((err) => {
        const subject = "定时任务BugStatisticsJob Error";
        logger.error(subject, err);
        void sendSystemEmailSafe({
          subject,
          text: getError(err).message,
        });
      });
    });
  }

  // 每天凌晨1点，做一次表格优化操作
  schedule.scheduleJob({ hour: 1, minute: 0 }, () => {
    jobOptimizeBugsTable().catch((err) => {
      const subject = "定时任务OptimizeBugsTableJob Error";
      logger.error(subject, err);
      void sendSystemEmailSafe({
        subject,
        text: getError(err).message,
      });
    });
  });

  // 每10分钟，清理下数据库表中的旧记录，避免数据量过大
  for (let i = 0; i < 60; i += 10) {
    schedule.scheduleJob({ minute: i }, () => {
      jobBugsLimitRecords().catch((err) => {
        const subject = "定时任务BugsLimitRecordsJob Error";
        logger.error(subject, err);
        void sendSystemEmailSafe({
          subject,
          text: getError(err).message,
        });
      });
    });
  }

  // 每10秒钟处理一次PPT任务状态
  for (let i = 0; i < 60; i += 10) {
    schedule.scheduleJob({ second: i }, () => {
      jobPPTTaskUpdateStatus().catch((err) => {
        const subject = "定时任务jobPPTTaskUpdateStatus Error";
        logger.error(subject, err);
        void sendSystemEmailSafe({
          subject,
          text: getError(err).message,
        });
      });
    });
  }

  // 每10分钟执行一次预约提醒任务
  for (let i = 0; i < 60; i += 10) {
    schedule.scheduleJob({ minute: i }, () => {
      jobBookingReminder().catch((err) => {
        const subject = "定时任务jobBookingReminder Error";
        logger.error(subject, err);
        void sendSystemEmailSafe({
          subject,
          text: getError(err).message,
        });
      });
    });
  }

  // 每10分钟执行一次告警检测任务
  for (let i = 0; i < 60; i += 10) {
    schedule.scheduleJob({ minute: i }, () => {
      jobBugsAlert().catch((err) => {
        const subject = "定时任务jobBugsAlert Error";
        logger.error(subject, err);
        void sendSystemEmailSafe({
          subject,
          text: getError(err).message,
        });
      });
    });
  }

  // 每天凌晨3点执行SSL证书状态刷新和自动续签
  schedule.scheduleJob({ hour: 3, minute: 0 }, () => {
    jobSSLCertificate().catch((err) => {
      const subject = "定时任务jobSSLCertificate Error";
      logger.error(subject, err);
      void sendSystemEmailSafe({
        subject,
        text: getError(err).message,
      });
    });
  });

  // 每天凌晨4点执行文章爬虫任务
  schedule.scheduleJob({ hour: 4, minute: 0 }, () => {
    runArticleCrawler().catch((err) => {
      const subject = "定时任务runArticleCrawler Error";
      logger.error(subject, err);
      void sendSystemEmailSafe({
        subject,
        text: getError(err).message,
      });
    });
  });

  // 每天凌晨5点执行sitemap.xml生成任务（文章爬虫在4点执行，错开时间）
  schedule.scheduleJob({ hour: 5, minute: 0 }, () => {
    jobGenerateSitemap().catch((err) => {
      const subject = "定时任务jobGenerateSitemap Error";
      logger.error(subject, err);
      void sendSystemEmailSafe({
        subject,
        text: getError(err).message,
      });
    });
  });
};

问题与改进方向

缺少分布式锁

问题：当前仅依赖 MySQL 行级锁（SELECT ... FOR UPDATE）进行并发控制，这在以下场景存在局限：

• 跨服务/跨进程的互斥操作（如定时任务抢锁、资源分配）
• 非数据库操作的临界区保护（如调用外部 API 前的去重检查）
• 高并发下的数据库行锁竞争可能导致死锁或性能瓶颈

改进方案：基于 Redis 实现分布式锁，提供标准化的锁接口。

// 建议：在 RedisUtils.mts 中新增分布式锁能力
import { randomUUID } from "node:crypto";

const LUA_ACQUIRE_LOCK = `
  -- 使用 SET NX + PX 原子化获取锁
  -- KEYS[1] = lockKey, ARGV[1] = requestId, ARGV[2] = ttl(ms)
  if redis.call("SET", KEYS[1], ARGV[1], "NX", "PX", ARGV[2]) then
    return 1
  end
  return 0
`;

const LUA_RELEASE_LOCK = `
  -- 使用 Lua 确保只有锁持有者才能释放（防止误删）
  -- KEYS[1] = lockKey, ARGV[1] = requestId
  if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
  end
  return 0
`;

export async function acquireLock(
  key: string,
  ttlMs = 30000,
): Promise<{ success: boolean; requestId: string }> {
  const client = getRedisClient();
  const requestId = randomUUID();
  const result = await client.eval(LUA_ACQUIRE_LOCK, {
    keys: [`distlock:${key}`],
    arguments: [requestId, String(ttlMs)],
  });
  return { success: result === 1, requestId };
}

export async function releaseLock(
  key: string,
  requestId: string,
): Promise<void> {
  const client = getRedisClient();
  await client.eval(LUA_RELEASE_LOCK, {
    keys: [`distlock:${key}`],
    arguments: [requestId],
  });
}

关键设计点：

• requestId（UUID）确保锁不会被其他实例误释放
• TTL 防止死锁（锁持有者崩溃后自动释放）
• Lua 脚本保证获取和释放的原子性

缺少分布式 ID 生成器

问题：当前数据库主键依赖 MySQL 自增 ID，会话 ID 使用 randomUUID()，缺少全局唯一、趋势递增的分布式 ID 生成器。自增 ID 在分库分表场景下会产生冲突，且不具备业务可读性。

改进方案：实现 Snowflake 风格的分布式 ID 生成器。

// 建议：新增 src/utils/IdGeneratorUtils.mts
const EPOCH = 1700000000000n; // 自定义起始时间戳
const WORKER_ID_BITS = 10n;
const SEQUENCE_BITS = 12n;

const MAX_WORKER_ID = (1n << WORKER_ID_BITS) - 1n;
const MAX_SEQUENCE = (1n << SEQUENCE_BITS) - 1n;

const WORKER_ID_SHIFT = SEQUENCE_BITS;
const TIMESTAMP_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS;

let lastTimestamp = -1n;
let sequence = 0n;

function tilNextMillis(lastTs: bigint): bigint {
  let ts = BigInt(Date.now());
  while (ts <= lastTs) {
    ts = BigInt(Date.now());
  }
  return ts;
}

export function generateId(workerId: number): bigint {
  const workerIdBig = BigInt(workerId);
  if (workerIdBig > MAX_WORKER_ID || workerIdBig < 0n) {
    throw new Error(`workerId 超出范围: 0-${MAX_WORKER_ID}`);
  }

  let timestamp = BigInt(Date.now()) - EPOCH;

  if (timestamp < lastTimestamp) {
    throw new Error("时钟回拨");
  }

  if (timestamp === lastTimestamp) {
    sequence = (sequence + 1n) & MAX_SEQUENCE;
    if (sequence === 0n) {
      timestamp = tilNextMillis(lastTimestamp);
    }
  } else {
    sequence = 0n;
  }

  lastTimestamp = timestamp;
  return (
    (timestamp << TIMESTAMP_SHIFT) | (workerIdBig << WORKER_ID_SHIFT) | sequence
  );
}

部署建议：通过环境变量 WORKER_ID 配置实例 ID，确保多实例间不重复。

定时任务重复执行

问题：schedules/index.mts 中的定时任务在应用进程内注册。如果部署多实例，所有实例会同时执行相同的任务，导致：

• 统计数据重复生成（如 Bugs 统计、表优化）
• 告警重复发送（如 SSL 证书到期提醒）
• 资源竞争（如文章爬虫重复抓取）

改进方案：使用 Redis SET NX 实现定时任务抢锁，只有获取到锁的实例才执行任务。

// 建议：封装定时任务锁工具
import { getRedisClient } from "#utils/RedisUtils";

export async function withScheduleLock<T>(
  jobName: string,
  task: () => Promise<T>,
  ttlSeconds = 300,
): Promise<T | null> {
  const client = getRedisClient();
  const lockKey = `schedule:lock:${jobName}`;
  const acquired = await client.set(lockKey, "1", {
    EX: ttlSeconds,
    NX: true,
  });

  if (!acquired) {
    // 其他实例已持有锁，跳过本次执行
    return null;
  }

  try {
    return await task();
  } finally {
    // 任务完成后释放锁（允许下一个周期重新抢锁）
    await client.del(lockKey);
  }
}

// 使用示例
import { withScheduleLock } from "#utils/ScheduleLockUtils";

// 在原定时任务回调中包裹
schedule.scheduleJob({ minute: i }, () => {
  withScheduleLock("bugs-statistics", () => jobBugsStatistics({ ... })).catch(...);
});

消费者幂等性缺失

问题：RabbitMQ 消费者在消息处理失败时，失败重放逻辑被注释掉（RabbitMQUtils.mts），且无幂等性保护机制，可能导致：

• 消息丢失（处理失败后既不 ack 也不 nack）
• 重复消费（消费者重启后重新消费已处理的消息）

改进方案：利用业务唯一键 + 去重表实现幂等消费。

// 建议：在消费者中增加幂等性检查
// 方案一：基于业务 ID 的去重表
async function processMessageIdempotent<T extends { id: number }>(
  message: T,
  processor: (msg: T) => Promise<void>,
): Promise<void> {
  // 使用唯一业务键检查是否已处理
  const processed = await knex("processed_message")
    .where({ message_id: message.id })
    .first();

  if (processed) {
    return; // 已处理，跳过
  }

  await processor(message);

  // 记录处理成功
  await knex("processed_message").insert({
    message_id: message.id,
    processed_at: knex.fn.now(),
  });
}

// 方案二：利用业务操作本身的唯一约束
// 在数据库中设置唯一索引（如 order_no、task_id），
// 重复插入时由数据库去重，避免重复处理

建议：优先使用方案二（数据库唯一约束），因为方案一需要额外维护去重表，且去重表本身也可能成为瓶颈。

缺少分布式追踪

问题：当前没有分布式追踪能力，当请求链路涉及多个服务或异步处理时，无法追踪完整的调用链，给问题排查带来困难。

改进方案：引入 OpenTelemetry 实现轻量级分布式追踪，透传 traceId。

// 建议：在中间件层注入 traceId
import { randomUUID } from "node:crypto";

// 在 app.mts 中新增 TraceMiddleware
export function traceMiddleware(
  req: ExpressRequest,
  _res: ExpressResponse,
  next: NextFunction,
) {
  const traceId =
    (req.headers["x-trace-id"] as string) || randomUUID().replace(/-/g, "");
  req.traceId = traceId;
  res.setHeader("x-trace-id", traceId);
  next();
}

// 在 RabbitMQ 消息中透传 traceId
interface RabbitMQMessageWithTrace<T> {
  traceId: string;
  data: T;
}

// 在消费者中提取 traceId 并注入日志上下文
logger.info(`[traceId=${traceId}] 处理消息`);

关键设计点：

• traceId 在 HTTP 请求入口生成，通过 x-trace-id 请求头透传
• 消息队列消费者从消息体中提取 traceId
• 日志输出始终携带 traceId，便于关联查询

缺少集群模式

问题：当前应用以单进程运行（bin/www.mts 使用 createServer），未使用 cluster 模块或 PM2，无法充分利用多核 CPU 资源。

改进方案：使用 PM2 启动多进程模式。

# 建议：在 package.json 中新增 PM2 配置
# ecosystem.config.mjs
export default {
  apps: [{
    name: "app",
    script: "dist/bin/www.mjs",
    instances: "max",          // 自动匹配 CPU 核心数
    exec_mode: "cluster",      // 集群模式
    max_memory_restart: "1G",  // 内存超过 1G 自动重启
    env: {
      NODE_ENV: "production",
      WORKER_ID: 0,            // 每个实例手动配置不同的 WORKER_ID
    },
  }],
};

注意事项：

• 启用 cluster 模式后需确保 trust proxy 配置正确，获取真实客户端 IP
• 应用内部状态（如内存缓存）需迁移到 Redis 等外部存储
• 定时任务需配合分布式锁使用（见上述改进方案）

缺少断路器

问题：外部服务调用（如 LLM API、第三方支付、CDN 资源获取）失败时，没有熔断保护机制，可能导致：

• 级联故障（一个外部服务不可用拖垮整个应用）
• 资源浪费（不断重试已经失败的服务）

改进方案：引入 opossum 断路器库。

// 建议：封装断路器工具
import CircuitBreaker from "opossum";

export function createServiceBreaker<T>(
  serviceName: string,
  action: (...args: unknown[]) => Promise<T>,
  options?: Partial<CircuitBreaker.Options>,
): CircuitBreaker<T> {
  const breaker = new CircuitBreaker(action, {
    timeout: 30000, // 30 秒超时
    errorThresholdPercentage: 50, // 50% 错误率触发熔断
    resetTimeout: 30000, // 30 秒后尝试半开
    name: serviceName,
    ...options,
  });

  breaker.on("open", () => logger.warn(`断路器打开: ${serviceName}`));
  breaker.on("halfOpen", () => logger.info(`断路器半开: ${serviceName}`));
  breaker.on("close", () => logger.info(`断路器关闭: ${serviceName}`));

  return breaker;
}

// 使用示例
const llmBreaker = createServiceBreaker("llm-api", chatWithLLM, {
  timeout: 60000, // LLM 调用超时设为 60 秒
});

总结

能力	当前状态	改进优先级
Redis 分布式缓存	✅ 已实现基础操作	—
分布式限流	✅ 滑动窗口 + 令牌桶	—
分布式会话	✅ 基于 Redis 存储	—
消息队列异步处理	✅ RabbitMQ 完整链路	中（幂等性）
MySQL 事务与行级锁	✅ 资金安全操作	—
HTTP 代理	✅ 支持开发/生产环境	—
WebSocket	✅ 基础通信	—
定时任务	✅ 单实例可用	高（多实例锁）
分布式锁	❌ 缺失	高
分布式 ID	❌ 缺失	中
分布式追踪	❌ 缺失	中
集群模式	❌ 单进程	中
断路器	❌ 缺失	低