后端性能优化实战：从诊断到优化的完整方法论

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

// Node.js性能分析
import { performance, PerformanceObserver } from 'perf_hooks';
import v8 from 'v8';

class PerformanceProfiler {
  private measurements = new Map<string, number[]>();

  // 测量函数执行时间
  async measure<T>(
    name: string,
    fn: () => Promise<T>
  ): Promise<T> {
    const start = performance.now();
    const heapBefore = v8.getHeapStatistics().used_heap_size;

    try {
      const result = await fn();

      const duration = performance.now() - start;
      const heapAfter = v8.getHeapStatistics().used_heap_size;
      const heapDelta = heapAfter - heapBefore;

      this.record(name, {
        duration,
        heapDelta,
        timestamp: Date.now()
      });

      return result;
    } catch (error) {
      const duration = performance.now() - start;
      this.record(name, {
        duration,
        error: true,
        timestamp: Date.now()
      });
      throw error;
    }
  }

  private record(name: string, metrics: any): void {
    if (!this.measurements.has(name)) {
      this.measurements.set(name, []);
    }
    this.measurements.get(name)!.push(metrics);
  }

  // 获取性能统计
  getStats(name: string): PerformanceStats {
    const measurements = this.measurements.get(name) || [];

    const durations = measurements.map(m => m.duration);
    const errors = measurements.filter(m => m.error).length;

    return {
      count: measurements.length,
      avgDuration: this.average(durations),
      minDuration: Math.min(...durations),
      maxDuration: Math.max(...durations),
      p95: this.percentile(durations, 95),
      p99: this.percentile(durations, 99),
      errorRate: errors / measurements.length
    };
  }

  private average(numbers: number[]): number {
    return numbers.reduce((a, b) => a + b, 0) / numbers.length;
  }

  private percentile(numbers: number[], p: number): number {
    const sorted = numbers.sort((a, b) => a - b);
    const index = Math.ceil((p / 100) * sorted.length) - 1;
    return sorted[index];
  }
}

// 使用示例
const profiler = new PerformanceProfiler();

app.get('/api/users/:id', async (request, response) => {
  const user = await profiler.measure('db.getUser', () =>
    db.getUser(request.params.id)
  );

  const posts = await profiler.measure('db.getUserPosts', () =>
    db.getUserPosts(request.params.id)
  );

  response.json({ user, posts });
});

// 定期输出统计
setInterval(() => {
  console.log('Performance Stats:');
  console.log('getUser:', profiler.getStats('db.getUser'));
  console.log('getUserPosts:', profiler.getStats('db.getUserPosts'));
}, 60000);

1
2
3
4
5
6
7
8

# 生成CPU火焰图
npm install -g 0x
0x --prof-flags="--record-coverage" node server.js

# 生成火焰图
npm install -g flamebearer
node --prof-process --preprocess - isolate-* isolate-*.log > v8.log
flamebearer v8.log > flamegraph.html

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

// 慢查询分析
class QueryOptimizer {
  // 分析查询计划
  async analyzeQuery(sql: string): Promise<QueryPlan> {
    const result = await this.db.query(`EXPLAIN ANALYZE ${sql}`);

    return {
      sql,
      plan: result.rows,
      totalCost: this.calculateTotalCost(result.rows),
      executionTime: this.getExecutionTime(result.rows)
    };
  }

  // 优化N+1查询
  async getUsersWithPosts(userIds: string[]): Promise<UserWithPosts[]> {
    // ❌ N+1查询 - 每个用户单独查询文章
    // const users = await db.getUsers(userIds);
    // for (const user of users) {
    //   user.posts = await db.getPostsByUser(user.id);  // N次查询
    // }

    // ✅ 使用IN查询 - 一次查询获取所有文章
    const users = await this.db.query(
      'SELECT * FROM users WHERE id = ANY($1)',
      [userIds]
    );

    const posts = await this.db.query(
      'SELECT * FROM posts WHERE user_id = ANY($1)',
      [userIds]
    );

    // 组装结果
    const postsByUser = this.groupBy(posts, 'user_id');

    return users.rows.map(user => ({
      ...user,
      posts: postsByUser[user.id] || []
    }));
  }

  // 使用JOIN优化
  async getUsersWithPostsJoin(userIds: string[]): Promise<UserWithPosts[]> {
    const result = await this.db.query(`
      SELECT
        u.id,
        u.name,
        u.email,
        json_agg(
          json_build_object(
            'id', p.id,
            'title', p.title,
            'created_at', p.created_at
          )
        ) as posts
      FROM users u
      LEFT JOIN posts p ON p.user_id = u.id
      WHERE u.id = ANY($1)
      GROUP BY u.id
    `, [userIds]);

    return result.rows;
  }

  private groupBy(array: any[], key: string): Record<string, any[]> {
    return array.reduce((result, item) => {
      const groupKey = item[key];
      if (!result[groupKey]) {
        result[groupKey] = [];
      }
      result[groupKey].push(item);
      return result;
    }, {} as Record<string, any[]>);
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

-- 创建复合索引（注意列顺序）
CREATE INDEX idx_users_email_verified ON users(email, verified_at);

-- 部分索引（只索引满足条件的行）
CREATE INDEX idx_active_users ON users(created_at)
WHERE status = 'active';

-- 表达式索引
CREATE INDEX idx_users_lower_email ON users(LOWER(email));

-- 覆盖索引（包含查询所需的所有列）
CREATE INDEX idx_posts_user_title_created
ON posts(user_id, title, created_at)
INCLUDE (content, status);

-- 分析索引使用情况
SELECT
  schemaname,
  tablename,
  indexname,
  idx_scan as index_scans,
  idx_tup_read as tuples_read,
  idx_tup_fetch as tuples_fetched
FROM pg_stat_user_indexes
WHERE schemaname = 'public'
ORDER BY idx_scan ASC;

-- 查找未使用的索引
SELECT
  schemaname,
  tablename,
  indexname,
  idx_scan
FROM pg_stat_user_indexes
WHERE idx_scan = 0
AND indexname NOT LIKE '%_pkey';

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

// PostgreSQL连接池配置
import { Pool, PoolConfig } from 'pg';

class DatabaseConnectionPool {
  private pool: Pool;

  constructor(config: PoolConfig) {
    this.pool = new Pool({
      ...config,
      // 连接池配置
      max: 20,                  // 最大连接数
      min: 5,                   // 最小连接数
      idleTimeoutMillis: 30000, // 空闲连接超时
      connectionTimeoutMillis: 10000, // 获取连接超时

      // 性能优化
      statement_timeout: 10000,  // 查询超时
      query_timeout: 10000,       // 查询超时

      // 连接验证
      verify: true
    });

    this.pool.on('error', (err) => {
      console.error('Unexpected error on idle client', err);
    });
  }

  async query<T>(sql: string, params?: any[]): Promise<T> {
    const client = await this.pool.connect();

    try {
      const start = Date.now();
      const result = await client.query(sql, params);
      const duration = Date.now() - start;

      // 记录慢查询
      if (duration > 1000) {
        console.warn('Slow query:', { sql, params, duration });
      }

      return result;
    } finally {
      client.release(); // 释放连接回池
    }
  }

  // 事务支持
  async transaction<T>(
    callback: (client: any) => Promise<T>
  ): Promise<T> {
    const client = await this.pool.connect();

    try {
      await client.query('BEGIN');
      const result = await callback(client);
      await client.query('COMMIT');
      return result;
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    } finally {
      client.release();
    }
  }

  // 优雅关闭
  async close(): Promise<void> {
    await this.pool.end();
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107

// 缓存层级
enum CacheLevel {
  L1 = 'L1', // 进程内缓存（最快，容量小）
  L2 = 'L2', // Redis缓存（较快，容量中等）
  L3 = 'L3'  // 数据库（最慢，容量大）
}

// 多级缓存实现
class MultiLevelCache {
  private l1Cache: LRUCache<string, any>;
  private l2Cache: Redis;

  constructor() {
    // L1: 内存缓存（1000条，5分钟过期）
    this.l1Cache = new LRUCache({
      max: 1000,
      ttl: 5 * 60 * 1000
    });

    // L2: Redis缓存（1小时过期）
    this.l2Cache = new Redis({
      host: 'localhost',
      port: 6379,
      defaultTTL: 60 * 60
    });
  }

  async get<T>(key: string): Promise<T | null> {
    // L1缓存查找
    const l1Value = this.l1Cache.get(key);
    if (l1Value !== undefined) {
      return l1Value;
    }

    // L2缓存查找
    const l2Value = await this.l2Cache.get(key);
    if (l2Value !== null) {
      // 回填L1缓存
      this.l1Cache.set(key, l2Value);
      return l2Value;
    }

    return null;
  }

  async set(key: string, value: any, ttl?: number): Promise<void> {
    // 同时写入L1和L2
    this.l1Cache.set(key, value);
    await this.l2Cache.set(key, value, ttl);
  }

  async invalidate(key: string): Promise<void> {
    // 同时使L1和L2失效
    this.l1Cache.delete(key);
    await this.l2Cache.del(key);
  }
}

// 缓存装饰器
function cacheable(ttl: number = 3600) {
  return function (
    target: any,
    propertyKey: string,
    descriptor: PropertyDescriptor
  ) {
    const originalMethod = descriptor.value;
    const cacheKeyPrefix = `${target.constructor.name}:${propertyKey}`;

    descriptor.value = async function (...args: any[]) {
      const cacheKey = `${cacheKeyPrefix}:${JSON.stringify(args)}`;

      // 尝试从缓存获取
      const cached = await this.cache.get(cacheKey);
      if (cached !== null) {
        return cached;
      }

      // 缓存未命中，执行原方法
      const result = await originalMethod.apply(this, args);

      // 写入缓存
      await this.cache.set(cacheKey, result, ttl);

      return result;
    };

    return descriptor;
  };
}

// 使用示例
class UserRepository {
  constructor(private cache: MultiLevelCache) {}

  @cacheable(600) // 缓存10分钟
  async getUserById(id: string): Promise<User> {
    return await this.db.query('SELECT * FROM users WHERE id = $1', [id]);
  }

  @cacheable(3600) // 缓存1小时
  async getUserPosts(userId: string): Promise<Post[]> {
    return await this.db.query(
      'SELECT * FROM posts WHERE user_id = $1',
      [userId]
    );
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

// Cache-Aside模式
class CacheAsidePattern {
  async get(key: string): Promise<any> {
    // 1. 查询缓存
    let value = await this.cache.get(key);

    // 2. 缓存未命中，查询数据库
    if (value === null) {
      value = await this.db.get(key);

      // 3. 写入缓存
      if (value !== null) {
        await this.cache.set(key, value);
      }
    }

    return value;
  }

  async set(key: string, value: any): Promise<void> {
    // 1. 更新数据库
    await this.db.set(key, value);

    // 2. 删除缓存（而非更新）
    await this.cache.del(key);
  }
}

// Write-Through模式
class WriteThroughPattern {
  async set(key: string, value: any): Promise<void> {
    // 同时写入缓存和数据库
    await Promise.all([
      this.cache.set(key, value),
      this.db.set(key, value)
    ]);
  }
}

// Write-Behind（异步写回）模式
class WriteBehindPattern {
  private writeQueue: AsyncQueue;

  constructor() {
    this.writeQueue = new AsyncQueue({
      concurrency: 10,
      timeout: 5000
    });
  }

  async set(key: string, value: any): Promise<void> {
    // 立即更新缓存
    await this.cache.set(key, value);

    // 异步更新数据库
    await this.writeQueue.add(async () => {
      await this.db.set(key, value);
    });
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

// 使用Worker Threads进行CPU密集型任务
import { Worker, isMainThread, parentPort, workerData } from 'worker_threads';

class WorkerPool {
  private workers: Worker[] = [];
  private queue: Array<{ task: any; resolve: Function; reject: Function }> = [];

  constructor(private size: number, private workerScript: string) {
    this.initialize();
  }

  private initialize(): void {
    for (let i = 0; i < this.size; i++) {
      const worker = new Worker(this.workerScript, {
        workerData: { id: i }
      });

      worker.on('message', (result) => {
        const { resolve } = this.queue.shift()!;
        resolve(result);
      });

      worker.on('error', (error) => {
        const { reject } = this.queue.shift()!;
        reject(error);
      });

      this.workers.push(worker);
    }
  }

  async execute<T>(task: any): Promise<T> {
    return new Promise((resolve, reject) => {
      this.queue.push({ task, resolve, reject });

      const availableWorker = this.workers.find(w => !this.isBusy(w));

      if (availableWorker) {
        availableWorker.postMessage(task);
      }
    });
  }

  private isBusy(worker: Worker): boolean {
    return worker.threadId !== 0;
  }
}

// Worker脚本 (heavy-task-worker.js)
if (!isMainThread) {
  parentPort?.on('message', (data) => {
    // 执行CPU密集型任务
    const result = heavyComputation(data);

    // 发送结果回主线程
    parentPort?.postMessage(result);
  });
}

function heavyComputation(data: any): any {
  // CPU密集型计算
  return data;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

// 并发控制
class ConcurrencyController {
  async executeAll<T, R>(
    items: T[],
    handler: (item: T) => Promise<R>,
    concurrency: number = 10
  ): Promise<R[]> {
    const results: R[] = [];
    const executing: Promise<void>[] = [];

    for (const item of items) {
      const promise = handler(item).then(result => {
        results.push(result);
        // 从执行列表中移除
        executing.splice(executing.indexOf(promise), 1);
      });

      executing.push(promise);

      // 控制并发数
      if (executing.length >= concurrency) {
        await Promise.race(executing);
      }
    }

    await Promise.all(executing);
    return results;
  }
}

// 使用示例
const controller = new ConcurrencyController();

const results = await controller.executeAll(
  userIds,
  async (userId) => {
    return await userService.getUser(userId);
  },
  5 // 同时最多5个请求
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

// 内存泄漏检测
import heapdump from 'heapdump';

class MemoryLeakDetector {
  private baseline: NodeJS.HeapSnapshot;

  // 记录基线快照
  takeBaseline(): void {
    const timestamp = new Date().toISOString();
    const filename = `baseline-${timestamp}.heapsnapshot`;

    heapdump.writeSnapshot(filename);
    console.log(`Baseline snapshot saved: ${filename}`);
  }

  // 记录当前快照
  takeSnapshot(): void {
    const timestamp = new Date().toISOString();
    const filename = `snapshot-${timestamp}.heapsnapshot`;

    heapdump.writeSnapshot(filename);
    console.log(`Snapshot saved: ${filename}`);
  }

  // 定期比较快照
  startMonitoring(interval: number = 60000): void {
    setInterval(() => {
      const usage = process.memoryUsage();

      console.log('Memory usage:', {
        rss: Math.round(usage.rss / 1024 / 1024) + ' MB',
        heapTotal: Math.round(usage.heapTotal / 1024 / 1024) + ' MB',
        heapUsed: Math.round(usage.heapUsed / 1024 / 1024) + ' MB',
        external: Math.round(usage.external / 1024 / 1024) + ' MB'
      });

      // 内存使用持续增长，可能存在泄漏
      if (usage.heapUsed > 500 * 1024 * 1024) { // 500MB
        console.warn('High memory usage detected, taking snapshot...');
        this.takeSnapshot();
      }
    }, interval);
  }
}

// 常见内存泄漏场景
class CommonLeaks {
  // ❌ 泄漏：全局变量
  globalData = []; // 持续增长

  // ✅ 正确：使用作用域变量
  getScopedData() {
    const data = [];
    return data;
  }

  // ❌ 泄漏：未清除的定时器
  startTimer() {
    setInterval(() => {
      this.doSomething();
    }, 1000);
  }

  // ✅ 正确：保存定时器引用并清理
  private timers: NodeJS.Timeout[] = [];

  startTimerFixed() {
    const timer = setInterval(() => {
      this.doSomething();
    }, 1000);

    this.timers.push(timer);
  }

  cleanup() {
    this.timers.forEach(timer => clearInterval(timer));
    this.timers = [];
  }

  // ❌ 泄漏：未清理的事件监听器
  subscribe() {
    emitter.on('data', this.handleData);
  }

  // ✅ 正确：清理事件监听器
  subscribeFixed() {
    this.handler = this.handleData.bind(this);
    emitter.on('data', this.handler);
  }

  unsubscribe() {
    emitter.off('data', this.handler);
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

// 流式处理大文件
import { createReadStream, createWriteStream } from 'fs';
import { pipeline } from 'stream/promises';
import { Transform } from 'stream';

class StreamProcessor {
  // 流式读取和处理
  async processLargeFile(
    inputPath: string,
    outputPath: string
  ): Promise<void> {
    const readStream = createReadStream(inputPath);
    const writeStream = createWriteStream(outputPath);

    // 转换流
    const transformStream = new Transform({
      transform(chunk: Buffer, encoding, callback) {
        // 处理数据块
        const processed = this.processChunk(chunk);
        callback(null, processed);
      }
    });

    // 使用pipeline自动处理错误和关闭流
    await pipeline(
      readStream,
      transformStream,
      writeStream
    );
  }

  private processChunk(chunk: Buffer): Buffer {
    // 处理逻辑
    return chunk;
  }

  // 分批处理数据库记录
  async *processBatch<T>(
    query: string,
    batchSize: number = 1000,
    processor: (batch: T[]) => Promise<void>
  ): AsyncGenerator<void> {
    let offset = 0;

    while (true) {
      const batch = await this.db.query(
        `${query} LIMIT $1 OFFSET $2`,
        [batchSize, offset]
      );

      if (batch.rows.length === 0) {
        break;
      }

      await processor(batch.rows);

      offset += batchSize;

      // 让出控制权，避免阻塞事件循环
      await new Promise(resolve => setImmediate(resolve));

      yield;
    }
  }
}