mysql - 如何避免nodejs mysql中大量查询出现死锁？

Question

我有很多 url，对于每个 url，我调用函数 load(url)，该函数解析 html，提取所需的数据并构建批量插入查询，如您在我的 test.js 代码中看到的那样。问题是，如果我有很多 url(比如 100+)，我会从 mysql 收到错误:ER_LOCK_DEADLOCK。我尝试使用 async.queue 但这在某种程度上不起作用(我不知道为什么，也许我使用的是错误的)。如何依次运行多个 url + 查询，避免并行执行(我认为这会导致死锁)？即使使用 async.queue 也会导致死锁(并非总是如此)。

测试.js

const request = require('request');
const async = require('async');
const pool = require('./database');

const urls = [
    'https://www.quora.com/What-is-the-best-way-to-have-delayed-job-queue-with-node-js',
    'https://de.wikipedia.org/wiki/Reinhardt-Zimmermann-L%C3%B6sung',
    'https://towardsdatascience.com/the-5-clustering-algorithms-data-scientists-need-to-know-a36d136ef68'
]

let load = function(url) {
    request({url: url}, function(error, response, html) {
        if(!error) {
            console.log(html);
            /**
             * 1. Parse HTML
             * 2. Create Array of Values
             * 3. Call pool.query(sql, [values], function(error) { ... })
             */
            let data = [{}];
            let sql = "INSERT IGNORE INTO tbl_test (title, content) VALUES ?";
            let values = [];

            data.forEach((item) => { values.push(item) });

            pool.query(sql, [values], function(error) { 
                if(error) throw error;
             })
        } else {
            console.log("handle error...");
        }
    })
}

let jobs = []

/*urls.forEach((url) => {
    //jobs.push(load(url)); // --> Works but fails if the urls list is to big -> mysql deadlock error!
    jobs.push(function(callback) { callback(load(url)) });
})*/

let q = async.queue(function(task, callback) {
    console.log("Task:", task.uri);
    callback();
})

q.drain = function() {
    console.log('all task completed');
    pool.end();
}

urls.forEach((url) => {
    q.push({uri: url}, function(err) {
        console.log('finished processing ...')
    });
});

数据库.js

require('dotenv').config();

const mysql = require('mysql');

let pool = mysql.createPool(
    {
        connectionLimit: 10,
        host: process.env.DB_HOST,
        port: process.env.DB_PORT,
        user: process.env.DB_USER,
        password: process.env.DB_PASSWORD,
        database: process.env.DB_NAME
    }
);

pool.getConnection((err, connection) => {
    if(err) {
        if(err.code === 'PROTOCOL_CONNECTION_LOST') {
            console.log('Database connection lost.')
        }

        if(err.code === 'ER_CON_COUNT_ERROR') {
            console.log('Database has too many connections.')
        }

        if(err.code === 'ECONNREFUSED') {
            console.log('Database connection refused.')
        }

        if(err.code === 'POOL_CLOSED') {
            console.log('Pool is closed.')
        }
    }

    if(connection) {
        connection.release()
    }

    return;
});

module.exports = pool;

Answer 1

我已更改代码以使用 async.series 而不是 async.queue，因为任务将在队列中并行运行(请参阅: https://caolan.github.io/async/docs.html#queue )。

测试.js

...
let tasks = [];

context.forEach((ctx) => {
    tasks.push(function(callback) { load(ctx, callback) });
});

async.series(tasks, function(err) {
    if(err) return next(err);
});