nodejs“并行”处理尝试

之前做过一些爬取方面的工作，由于node不能多线程，为了提高抓取效率，都是使用child_process.fork来多进程跑任务，然后通过message事件与主进程进行通信，代码编写的时候都是用的yield/await之类的同步写法，于是这次尝试利用node非阻塞I/O的机制，利用多个函数同时运行来模拟多线程，效果如何呢？

尝试“并行”发送HTTP请求

server.js

用来统计qps，将产出的数据status.txt里的内容复制到echarts的官方示例里进行可视化，从而验证是否能达到“并行”的效果

const fs = require('fs');
const Koa = require('koa');
const app = new Koa();

// 用来统计qps
let last_time = new Date(),
    init_timestamp = last_time.getTime(),
    count = 0;

// 运行时长60s
let run_secs = 60;

// 用来存储qps历史，用于绘制曲线图
let qps_list = [],
    timestr_list = [];

app.use(async ctx => {
    // 简单的模拟计算qps
    let cur_time = new Date(),
        cur_timestr = cur_time.toLocaleTimeString(),
        cur_timestamp = cur_time.getTime(),
        last_timestr = last_time.toLocaleTimeString();

    if (cur_timestr !== last_timestr) {
        let timestamp_cost = Math.round((cur_timestamp - init_timestamp) / 1000);

        console.log(`\n${cur_timestr}: ${timestamp_cost} qps*********************************`);
        console.log(count);

        qps_list.push(count);
        timestr_list.push(cur_timestr);

        if (timestamp_cost >= run_secs) {
            // 将运行结果存储起来，打开http://echarts.baidu.com/examples/editor.html?c=line-smooth，复制内容查看曲线图
            let option_str = JSON.stringify({
                tooltip: {
                    trigger: 'axis'
                },
                xAxis: {
                    type: 'category',
                    data: timestr_list
                },
                yAxis: {
                    type: 'value'
                },
                series: [{
                    data: qps_list,
                    type: 'line',
                    smooth: true
                }]
            }, null, 2);
            fs.writeFileSync('./status.txt', `option=${option_str}`);

            console.log('1.复制status.txt的内容');
            console.log('2.打开http://echarts.baidu.com/examples/editor.html?c=line-smooth');
            console.log('3.粘贴在左边代码区域');
            console.log('4.点击"运行"，在右侧区域查看');
            process.exit();
        }

        last_time = cur_time;
        count = 1;
    } else {
        count++;
    }

    // 模拟服务端处理请求的时间
    await delay();

    ctx.body = 'hello';
});

function delay () {
    return new Promise((resolve) => {
        setTimeout(resolve, 250);
    });
}

app.listen(3000);
复制代码

单进程版本

client.js

const axios = require('axios');

async function sendRequest (id) {
    return new Promise((resolve, reject) => {
        axios.get(`http://localhost:3000?id=${id}`).then(res => {
            resolve(res.data);
        }).catch(e => {
            reject(e);
        });
    });
}

function run () {
    let threads = 1;

    for (let i = 0; i < threads; i++) {
        makeThread(i);
    }
}

async function makeThread (id) {
    while (true) {
        try {
            await sendRequest(id);
        } catch (e) {
            console.log(id, e.message);
            process.exit();
        }
    }
}

run();
复制代码

多进程版本（进行对照）

client_center.js

const fork = require('child_process').fork;

function run () {
    let threads = 1;

    for (let i = 0; i < threads; i++) {
        fork('./client_worker.js', [i]);
    }
}

run();
复制代码

client_worker.js

const axios = require('axios');

let id = process.argv[2];

async function sendRequest () {
    return new Promise((resolve, reject) => {
        axios.get(`http://localhost:3000?id=${id}`).then(res => {
            resolve(res.data);
        }).catch(e => {
            reject(e);
        });
    });
}

async function makeThread () {
    while (true) {
        try {
            await sendRequest();
        } catch (e) {
            console.log(id, e.message);
            process.exit();
        }
    }
}

makeThread();
复制代码

运行结果（服务设置延时250ms）

2核机器

threads	单进程版本	多进程版本	备注
1			区别不大
5			区别不大
50			多进程效果弱于单进程版本
100			多进程效果弱于单进程版本
200			多进程弱于单进程版本，且多进程版本总是报错：read ECONNRESET/connect ECONNRESET/socket hang up
300			多进程弱于单进程版本，且多进程版本总是报错：read ECONNRESET/connect ECONNRESET/socket hang up

观察

对比结果让我挺吃惊的，这样看来单进程的模拟效果居然会比多进程好，但突然想到自己电脑上才几核，怎么同时跑几百个进程....... 登录到公司服务器上（48核）继续实验：

48核机器

threads	单进程版本	多进程版本	备注
30			区别不大
40			区别不大
100			qps峰值相同，但多进程更稳定
200			多进程版本优于单进程版本
300			多进程版本优于单进程版本
1000			多进程版本优于单进程版本
1500			多进程版本优于单进程版本，但threads增加所带来的收益较低，多进程版本峰值4318<1500*4，单进程版本峰值3058<1500*4
3000			单进程版本(峰值2969)优于多进程版本(峰值1500)

观察

• 确实能通过多个函数同时运行的方式来模拟多线程的效果
• 当threads设置与核数差距不大时，两者效果差不多。
• 在高性能机器上， 在一定范围(大部分范围)内 ，threads越大，多进程版本的效果越好，但超过这个范围(极端情况)，单进程版本反而表现突出
• 在低性能机器上，单进程版本表现更好，由于出现的read ECONNRESET/connect ECONNRESET/socket hang up等错误使得无法继续增大threads数量观察下去
• 低性能机器上两个版本都会表现出奇怪的周期性，在高性能机器上多进程版本会更稳定一些

分析

• client.js能模拟“并行”的效果实际上是利用网络耗时远大于代码循环的原理，第一次for循环连续发送threads个网络请求，然后在处理回调的时候又发送新的网络请求，效果就变成了多个线程在不停的发请求。

不负责任的猜测

• 高/低性能表现不一致，低性能机器是mac，libuv中使用kqueue处理网络I/O，高性能机器时linux，libuv中使用epoll处理

新的问题

• 该服务性能的极限QPS是多少
• 奇怪的曲线产生原因