名称： senior-backend
描述： 当用户提出以下需求时，应使用此技能："设计 REST API"、"优化数据库查询"、"实现身份验证"、"构建微服务"、"评审后端代码"、"设置 GraphQL"、"处理数据库迁移"或"对 API 进行负载测试"。适用于 Node.js/Express/Fastify 开发、PostgreSQL 优化、API 安全和后端架构模式。

高级后端工程师

后端开发模式、API 设计、数据库优化与安全实践。

目录

• 快速开始
• 工具概览
• API 脚手架生成器
• 数据库迁移工具
• API 负载测试工具
• 后端开发工作流
• API 设计工作流
• 数据库优化工作流
• 安全加固工作流
• 参考文档
• 常用模式速查

快速开始

# 根据 OpenAPI 规范生成 API 路由
python scripts/api_scaffolder.py openapi.yaml --framework express --output src/routes/

# 分析数据库架构并生成迁移文件
python scripts/database_migration_tool.py --connection postgres://localhost/mydb --analyze

# 对 API 端点进行负载测试
python scripts/api_load_tester.py https://api.example.com/users --concurrency 50 --duration 30

工具概览

1. API 脚手架生成器

根据模式定义生成 API 路由处理器、中间件和 OpenAPI 规范。

输入： OpenAPI 规范 (YAML/JSON) 或数据库架构
输出： 路由处理器、验证中间件、TypeScript 类型

用法：

# 根据 OpenAPI 规范生成 Express 路由
python scripts/api_scaffolder.py openapi.yaml --framework express --output src/routes/

# 输出示例：
# 已在 src/routes/ 中生成 12 个路由处理器
# - GET /users (listUsers)
# - POST /users (createUser)
# - GET /users/{id} (getUser)
# - PUT /users/{id} (updateUser)
# - DELETE /users/{id} (deleteUser)
# ...
# 已创建验证中间件：src/middleware/validators.ts
# 已创建 TypeScript 类型：src/types/api.ts

# 根据数据库架构生成
python scripts/api_scaffolder.py --from-db postgres://localhost/mydb --output src/routes/

# 根据现有路由生成 OpenAPI 规范
python scripts/api_scaffolder.py src/routes/ --generate-spec --output openapi.yaml

支持的框架：
- Express.js (--framework express)
- Fastify (--framework fastify)
- Koa (--framework koa)

2. 数据库迁移工具

分析数据库架构，检测变更，并生成支持回滚的迁移文件。

输入： 数据库连接字符串或架构文件
输出： 迁移文件、架构差异报告、优化建议

用法：

# 分析当前架构并提供优化建议
python scripts/database_migration_tool.py --connection postgres://localhost/mydb --analyze

# 输出示例：
# === 数据库分析报告 ===
# 表数量：24
# 总行数：1,247,832
#
# 缺失索引 (发现 5 处)：
#   orders.user_id - 平均查询时间 847ms，建议添加索引
#   products.category_id - 平均查询时间 234ms，建议添加索引
#
# N+1 查询风险 (发现 3 处)：
#   users -> orders 关联关系 (未启用预加载)
#
# 建议的迁移：
#   1. 为 orders(user_id) 添加索引
#   2. 为 products(category_id) 添加索引
#   3. 为 order_items(order_id, product_id) 添加复合索引

# 根据架构差异生成迁移
python scripts/database_migration_tool.py --connection postgres://localhost/mydb \
  --compare schema/v2.sql --output migrations/

# 输出示例：
# 已生成迁移文件：migrations/20240115_add_user_indexes.sql
# 已生成回滚文件：migrations/20240115_add_user_indexes_rollback.sql

# 模拟运行迁移 (dry-run)
python scripts/database_migration_tool.py --connection postgres://localhost/mydb \
  --migrate migrations/20240115_add_user_indexes.sql --dry-run

3. API 负载测试工具

执行可配置并发数的 HTTP 负载测试，测量延迟百分位数和吞吐量。

输入： API 端点 URL 和测试配置
输出： 包含延迟分布、错误率、吞吐量指标的性能报告

用法：

# 基础负载测试
python scripts/api_load_tester.py https://api.example.com/users --concurrency 50 --duration 30

# 输出示例：
# === 负载测试结果 ===
# 目标：https://api.example.com/users
# 持续时间：30s | 并发数：50
#
# 吞吐量：
#   总请求数：15,247
#   请求数/秒：508.2
#   成功：15,102 (99.0%)
#   失败：145 (1.0%)
#
# 延迟 (ms)：
#   最小值：12
#   平均值：89
#   P50：67
#   P95：198
#   P99：423
#   最大值：1,247
#
# 错误：
#   连接超时：89
#   HTTP 503：56
#
# 建议：P99 延迟 (423ms) 超过 200ms 目标。
# 可考虑：连接池、查询优化或水平扩展。

# 使用自定义请求头和请求体进行测试
python scripts/api_load_tester.py https://api.example.com/orders \
  --method POST \
  --header "Authorization: Bearer token123" \
  --body '{"product_id": 1, "quantity": 2}' \
  --concurrency 100 \
  --duration 60

# 比较两个端点
python scripts/api_load_tester.py https://api.example.com/v1/users https://api.example.com/v2/users \
  --compare --concurrency 50 --duration 30

后端开发工作流

API 设计工作流

适用于设计新 API 或重构现有端点。

步骤 1：定义资源和操作

# openapi.yaml
openapi: 3.0.3
info:
  title: 用户服务 API
  version: 1.0.0
paths:
  /users:
    get:
      summary: 列出用户
      parameters:
        - name: limit
          in: query
          schema:
            type: integer
            default: 20
    post:
      summary: 创建用户
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/CreateUser'

步骤 2：生成路由脚手架

python scripts/api_scaffolder.py openapi.yaml --framework express --output src/routes/

步骤 3：实现业务逻辑

// src/routes/users.ts (生成后自定义)
export const createUser = async (req: Request, res: Response) => {
  const { email, name } = req.body;

  // 添加业务逻辑
  const user = await userService.create({ email, name });

  res.status(201).json(user);
};

步骤 4：添加验证中间件

# 验证中间件根据 OpenAPI 模式自动生成
# src/middleware/validators.ts 包含：
# - 请求体验证
# - 查询参数验证
# - 路径参数验证

步骤 5：生成更新的 OpenAPI 规范

python scripts/api_scaffolder.py src/routes/ --generate-spec --output openapi.yaml

数据库优化工作流

适用于查询缓慢或需要提升数据库性能的场景。

步骤 1：分析当前性能

python scripts/database_migration_tool.py --connection $DATABASE_URL --analyze

步骤 2：识别慢查询

-- 检查查询执行计划
EXPLAIN ANALYZE SELECT * FROM orders
WHERE user_id = 123
ORDER BY created_at DESC
LIMIT 10;

-- 关注点：Seq Scan (不佳)，Index Scan (良好)

步骤 3：生成索引迁移

python scripts/database_migration_tool.py --connection $DATABASE_URL \
  --suggest-indexes --output migrations/

步骤 4：测试迁移 (模拟运行)

python scripts/database_migration_tool.py --connection $DATABASE_URL \
  --migrate migrations/add_indexes.sql --dry-run

步骤 5：应用并验证

# 应用迁移
python scripts/database_migration_tool.py --connection $DATABASE_URL \
  --migrate migrations/add_indexes.sql

# 验证改进效果
python scripts/database_migration_tool.py --connection $DATABASE_URL --analyze

安全加固工作流

适用于准备将 API 投入生产或进行安全审查后。

步骤 1：审查身份验证设置

// 验证 JWT 配置
const jwtConfig = {
  secret: process.env.JWT_SECRET,  // 必须来自环境变量，切勿硬编码
  expiresIn: '1h',                 // 短期令牌
  algorithm: 'RS256'               // 推荐使用非对称加密
};

步骤 2：添加速率限制

import rateLimit from 'express-rate-limit';

const apiLimiter = rateLimit({
  windowMs: 15 * 60 * 1000,  // 15 分钟
  max: 100,                   // 每个窗口 100 个请求
  standardHeaders: true,
  legacyHeaders: false,
});

app.use('/api/', apiLimiter);

步骤 3：验证所有输入

import { z } from 'zod';

const CreateUserSchema = z.object({
  email: z.string().email().max(255),
  name: z.string().min(1).max(100),
  age: z.number().int().positive().optional()
});

// 在路由处理器中使用
const data = CreateUserSchema.parse(req.body);

步骤 4：使用攻击模式进行负载测试

# 测试速率限制
python scripts/api_load_tester.py https://api.example.com/login \
  --concurrency 200 --duration 10 --expect-rate-limit

# 测试输入验证
python scripts/api_load_tester.py https://api.example.com/users \
  --method POST \
  --body '{"email": "not-an-email"}' \
  --expect-status 400

步骤 5：审查安全头

import helmet from 'helmet';

app.use(helmet({
  contentSecurityPolicy: true,
  crossOriginEmbedderPolicy: true,
  crossOriginOpenerPolicy: true,
  crossOriginResourcePolicy: true,
  hsts: { maxAge: 31536000, includeSubDomains: true },
}));

参考文档

常用模式速查

REST API 响应格式

{
  "data": { "id": 1, "name": "John" },
  "meta": { "requestId": "abc-123" }
}

错误响应格式

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "邮箱格式无效",
    "details": [{ "field": "email", "message": "必须是有效的邮箱地址" }]
  },
  "meta": { "requestId": "abc-123" }
}

HTTP 状态码

数据库索引策略

-- 单列索引 (用于等值查找)
CREATE INDEX idx_users_email ON users(email);

-- 复合索引 (用于多列查询)
CREATE INDEX idx_orders_user_status ON orders(user_id, status);

-- 部分索引 (用于过滤查询)
CREATE INDEX idx_orders_active ON orders(created_at) WHERE status = 'active';

-- 覆盖索引 (避免回表查询)
CREATE INDEX idx_users_email_name ON users(email) INCLUDE (name);

常用命令

# API 开发
python scripts/api_scaffolder.py openapi.yaml --framework express
python scripts/api_scaffolder.py src/routes/ --generate-spec

# 数据库操作
python scripts/database_migration_tool.py --connection $DATABASE_URL --analyze
python scripts/database_migration_tool.py --connection $DATABASE_URL --migrate file.sql

# 性能测试
python scripts/api_load_tester.py https://api.example.com/endpoint --concurrency 50
python scripts/api_load_tester.py https://api.example.com/endpoint --compare baseline.json