IO/CPU密集型优化思路—池化技术

场景1：数据库连接池（IO密集型）

如果每次数据库请求都请求都开启一个连接，然后用完销毁，性能会有问题。

分析程序的日志之后，你发现系统慢的原因出现在和数据库的交互上。因为你们数据库的调用方式是先获取数据库的连接，然后依靠这条连接从数据库中查询数据，最后关闭连接释放数据库资源。

我们统计了一段时间的 SQL 执行时间，发现 SQL 的平均执行时间大概是 1ms，也就是说相比于 SQL 的执行，MySQL 建立连接的过程是比较耗时的。

连接池设计

数据库连接池有两个最重要的配置：最小连接数和最大连接数。

  • 如果当前连接数小于最小连接数，则创建新的连接处理数据库请求；
  • 如果连接池中有空闲连接则复用空闲连接；
  • 如果空闲池中没有连接并且当前连接数小于最大连接数，则创建新的连接处理请求；
  • 如果当前连接数已经大于等于最大连接数，则按照配置中设定的时间（C3P0 的连接池配置是 checkoutTimeout）等待旧的连接可用；
  • 如果等待超过了这个设定时间则向用户抛出错误。

特别注意第一条，当所有连接小于最小连接时，优先创建连接，而不是先去空闲池中获取连接。

对于数据库连接池，根据我的经验，一般在线上我建议最小连接数控制在 10 左右，最大连接数控制在 20～30 左右即可。

除此之外，连接放在池中，分为空闲连接池和忙碌连接池：

  • 空闲连接池：其中的连接，可以直接取出，并返回给开发者
  • 忙碌连接池：从空闲连接池取出的就放入忙碌连接池

要检测mysql连接的状态，原因有是：MySQL 有个参数是“wait_timeout”，控制着当数据库连接闲置多长时间后，数据库会主动地关闭这条连接。这个机制对于数据库使用方是无感知的，所以当我们使用这个被关闭的连接时就会发生错误。

检测方法：启动一个线程来定期检测连接池中的连接是否可用，比如使用连接发送“select 1”的命令给数据库看是否会抛出异常，如果抛出异常则将这个连接从连接池中移除，并且尝试关闭。目前 C3P0 连接池可以采用这种方式来检测连接是否可用，也是我比较推荐的方式。

代码实现

类的定义：

1
2
3
4
5
6
7
8
9
10
11
class DbConnectPool {
    constructor() {
        this.minCount = 10;
        this.maxCount = 30;

        this.freePool = [];
        this.busyPool = [];

        this.checker = null;
    }
 }

获取连接和回收连接逻辑：

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
/**
 * 新建数据库连接（代码省略）
 */
async createConnect() {}

/**
 * 获取可用的数据库连接
 */
async getConnect(retryTimes = 0) {
    if (retryTimes > 2) {
        throw new Error('暂无可用数据库连接')
    }

    if (this.freePool.length + this.busyPool.length < this.minCount) {
        const n = await this.createConnect();
        this.busyPool.push(n);
        return n;
    }

    if (this.freePool.length > 0) {
        const top = this.pool.pop();
        this.busyPool.push(top);
        return top;
    }

    if (this.freePool.length + this.busyPool.length < this.maxCount) {
        const n = await this.createConnect();
        this.busyPool.push(n);
        return n;
    }

    await sleep(10);
    return await this.getConnect(retryTimes + 1)
}
/**
 * 回收连接
 */
recycleConnect(connect) {
    const index = this.busyPool.findIndex(item => item === connect)
    if (index === -1) {
        return;
    }
    this.busyPool.splice(index, 1);
    this.freePool.push(connect);
}

心跳策略检查连接可用性逻辑：

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
startCheck() {
    if (this.checker) {
        return;
    }

    this.checker = setInterval(() => {
        this.checkConnect(this.busyPool, 'busy');
        this.checkConnect(this.freePool, 'free')
    }, 100)
}

async checkConnect(allConnect, mode = '') {
    const validConnect = []
    for (let i = 0; i < allConnect; ++i) {
        try {
            await connect.send('SELECT 1');
            validConnect.push(allConnect[i])
        } catch (error) {
            console.log('......')
        }
    }
    if (mode === 'busy') {
        this.busyPool = validConnect;
    } else if (mode === 'free') {
        this.freePool = validConnect;
    }
}

销毁逻辑，防止内存泄漏：

1
2
3
4
5
6
7
8
9
10
async destory() {
    const allConnect = [...this.busyPool, ...this.freePool];
    for (const connect of allConnect) {
        await connect.close();
    }
    this.busyPool.length = 0;
    this.freePool.length = 0;

    clearInterval(this.checker);
}

场景2：计算线程池（计算密集型）

设计

JDK 1.5 中引入的 ThreadPoolExecutor 就是一种线程池的实现。

逻辑：

  • 如果线程池中的线程数少于 coreThreadCount 时，处理新的任务时会创建新的线程；
  • 如果线程数大于 coreThreadCount 则把任务丢到一个队列里面，由当前空闲的线程执行；
  • 当队列中的任务堆积满了的时候，则继续创建线程，直到达到 maxThreadCount；
  • 当线程数达到 maxTheadCount 时还有新的任务提交，那么我们就不得不将它们丢弃了。

Untitled.png

为什么超过minCount（coreCount）后，先插入队列，而不是像连接池直接创建？

  1. JDK 实现的这个线程池优先把任务放入队列暂存起来，而不是创建更多的线程，它比较适用于执行 CPU 密集型的任务，也就是需要执行大量 CPU 运算的任务。
    因为执行 CPU 密集型的任务时 CPU 比较繁忙，因此只需要创建和 CPU 核数相当的线程就好了，多了反而会造成线程上下文切换，降低任务执行效率。所以当前线程数超过核心线程数时，线程池不会增加线程，而是放在队列里等待核心线程空闲下来。
  2. 针对IO密集型操作，比如缓存查询、数据库查询，在执行 IO 操作的时候 CPU 就空闲了下来，这时如果增加执行任务的线程数而不是把任务暂存在队列中，就可以在单位时间内执行更多的任务，大大提高了任务执行的吞吐量。