javascript - Nodejs 卡住 MySQL 大量结果输入 REDIS HMSET

Question

我有一个脚本，它使用 SELECT 对 MySQL 执行 2 个查询，结果查询每个产生一个 226,393 行的结果集和另一个 529,976 行的结果集。当输入 REDIS 时，每行添加 4 次，每个要存储的唯一数据一个。

第一个查询似乎完成了，没问题，但第二个查询似乎停在了中间并停留在那里。我是 NODEjs 和 REDIS 的新手，并且有传统的 MySQL 背景。

代码如下:

// Start the server
http.createServer(function(req, res){
    console.log('Request received');

    for(var n = 0; n < qryArray.length; n++) {
        var qry = qryArray[n];

        // Send the query
        //console.log( n + ' :: ' + qry);
        connection.query(qry, function(err, rows, fields){
            if(err){
                console.log(err);
            }else{
                console.log('Query response ' + rows.length + ' rows');
                //console.log(util.inspect(process.memoryUsage()));

                errorCount = 0;
                goodCount  = 0;
                for(var i = 0; i < rows.length; i++){
                    var row = rows[i];

                    j = 0;
                    //var fields = result.fields.map(function(f) { return f.name; })
                    fields.forEach(function(f){
                        if(f.name != 'latlng' && f.name != 'market' && f.name != 'technology'){
                            j++;

                            //console.log(('' + row[f.name]).toUpperCase() + '~' + i + '-' + j + '~' + row.latlng);
                            redisClient.hmset(('' + row[f.name]).toUpperCase() + '~' + i + '-' + j + '~' + row.latlng, row, function(error, result){
                                if(error){
                                    errorCount++;
                                    console.log(errorCount + ' ' + error + ' ' + f.name.toUpperCase() + ' : ' + ('' + row[f.name]).toUpperCase() + '~' + i + '-' + j + '~' + row.latlng + ' :: ' + JSON.stringify(row));
                                }else{
                                    goodCount++;
                                    console.log(goodCount);
                                    //console.log(goodCount + ' Redis stored: ' + f.name.toUpperCase() + ' : ' + ('' + row[f.name]).toUpperCase() + ' ' + result);
                                }
                            });
                        }

                    });
                    //goodCount++;
                    //console.log(goodCount);
                }
                console.log('Finished indexing ' + rows.length + ' rows ' + goodCount);
            }
        });
    }
    console.log('Queries processed');

}).listen(port);
console.log('Server running on port ' + port);

有人可以指出我哪里出了问题，或者如何针对大型数据集进行改进吗？谢谢!

在获取有关添加异步库以使用队列的指针后，我修改了代码，但得到了一个令人讨厌的错误:

{ [Error: Connection lost: The server closed the connection.] fatal: true, code: 'PROTOCOL_CONNECTION_LOST' }

这是添加了功能的新代码，请帮忙:

// Initialize the queue
var q = async.queue(function(task){

    //console.log(('' + row[f.name]).toUpperCase() + '~' + i + '-' + j + '~' + row.latlng);
    redisClient.hmset(task.hk, task.r, function(error, result){
        if(error){
            errorCount++;
            console.log('Error: ' + errorCount);
            //console.log(errorCount + ' ' + error + ' ' + f.name.toUpperCase() + ' : ' + ('' + row[f.name]).toUpperCase() + '~' + i + '-' + j + '~' + row.latlng + ' :: ' + JSON.stringify(row));
        }else{
            goodCount++;
            console.log('Good: ' + goodCount);
            //console.log(goodCount + ' Redis stored: ' + f.name.toUpperCase() + ' : ' + ('' + row[f.name]).toUpperCase() + ' ' + result);
        }
    });

}, 50000);

// Assign callback for when all items in queue have been processed
q.drain = function(){
    console.log('All queue items have been processed ' + goodCount);
}

// Start the server
http.createServer( function( req, res ){
    console.log('Request received');

    for(var n = 0; n < qryArray.length; n++) {
        var qry = qryArray[n];

        // Send the query
        //console.log( n + ' :: ' + qry);
        connection.query(qry, function(err, rows, fields){
            if(err){
                console.log(err);
            }else{
                console.log('Query response ' + rows.length + ' rows');
                //console.log(util.inspect(process.memoryUsage()));

                errorCount = 0;
                goodCount  = 0;
                for(var i = 0; i < rows.length; i++){
                    var row = rows[i];

                    var j = 0;
                    //var fields = result.fields.map(function(f) { return f.name; })
                    fields.forEach(function(f){
                        if(f.name != 'latlng' && f.name != 'market' && f.name != 'technology'){
                            j++;

                            var hkey = ('' + row[f.name]).toUpperCase() + '~' + i + '-' + j + '~' + row.latlng;
                            var task = {r: row, hk: hkey};
                            q.push(task, function(err){
                                if(err) console.log(err);
                            });
                        }
                    });

                }
                console.log('Finished indexing ' + rows.length + ' rows');
            }
        });
    }
    console.log('Queries processed');

}).listen(port);
console.log('Server running on port ' + port);

错误发生在 50K 队列项目成功完成后(队列已初始化为 50k 并发项目)。

断线问题改成MySQL pool机制，还是断线，求助。此外，在实际插入到 REDIS 之前需要花费大量时间，那里也有问题吗？

// Start the server
http.createServer(function(req, res){
    res.writeHead(200);
    res.end();

    console.log('Request received');
    pool.getConnection(function(err, connection){
        if(err){
            connection.release();
            return console.log('Database connection error ' + err);
        }
        console.log('Database is connected ' + connection.threadId + ' ...');

        for(var n = 0; n < qryArray.length; n++){
            var qry = qryArray[n];

            // Send the query
            connection.query(qry, function(err, rows, fields){
                connection.release();
                if(err){
                    return console.log('Query error: ' + err);
                }
                console.log('Query response ' + rows.length + ' rows');

                errorCount = 0;
                goodCount  = 0;
                for(var i = 0; i < rows.length; i++){
                    var row = rows[i];

                    var j = 0;
                    fields.forEach(function(f){
                        if(f.name != 'latlng' && f.name != 'market' && f.name != 'technology'){
                            j++;

                            var hkey = ('' + row[f.name]).toUpperCase() + '~' + i + '-' + j + '~' + row.latlng;
                            var task = {r: row, hk: hkey};
                            q.push(task, function(err){
                                if(err) console.log(err);
                            });
                        }
                    });
                }
                console.log('Finished indexing ' + rows.length + ' rows');
            });

            connection.on('error', function(err){
                return console.log('Database connection error ' + err);
            });
        }
        console.log('Queries processed');
    });

}).listen(port);
console.log('Server running on port ' + port);

Answer 1

谢谢大家帮助我，脚本现在可以使用 MySQL 池、作业队列，我添加了 HTTP 调度程序以获取有关主要任务进度的简单状态报告:将 MySQL 数据集结果插入 REDIS。花了几个小时才完成(+300 万次插入)。

我发布了生成的脚本，希望它既可以改进，也可以帮助启发处于类似情况的其他人。

// MySQL initialization
var pool = mysql.createPool({
    connectionLimit : 10,
    host     : 'localhost',
    user     : 'gta_ro',
    password : 'glacier',
    database : 'tower'
});

// Async Queue initialization
var insertCount = 0;
var q = async.queue(function(task, callback){

    redisClient.hmset(task.hk, task.r, function(error, result){
        if(error){
            errorCount++;
            console.log('Error: ' + errorCount);
            //console.log(errorCount + ' ' + error + ' ' + f.name.toUpperCase() + ' : ' + ('' + row[f.name]).toUpperCase() + '~' + i + '-' + j + '~' +     row.latlng + ' :: ' + JSON.stringify(row));
        }else{
            goodCount++;
            console.log('Good: ' + goodCount);
            //console.log(goodCount + ' Redis stored: ' + f.name.toUpperCase() + ' : ' + ('' + row[f.name]).toUpperCase() + ' ' + result);
        }
    });
    return setImmediate(function() { callback() });
}, 1000);

// Async Queue Empty callback (triggered after last item has been consumed)
q.drain = function(){
    console.log('All queue items have been processed ' + goodCount);
}

// Start the server
httpStarted = false;
http.createServer(function(req, res){
    dispatcher.dispatch(req, res);

    if(!httpStarted){
        httpStarted = true;

        console.log('Server running on port ' + port);
        pool.getConnection(function(err, connection){
            if(err){
                connection.release();
                return console.log('Database connection error ' + err);
            }
            console.log('Database is connected ' + connection.threadId + ' ...');

            for(var n = 0; n < qryArray.length; n++){
                var qry = qryArray[n];

                // Send the query
                connection.query(qry, function(err, rows, fields){
                    if(err){
                        return console.log('Query error: ' + err);
                    }

                    //connection.release();
                    console.log('Query response ' + rows.length + ' rows');

                    errorCount = 0;
                    goodCount  = 0;
                    for(var i = 0; i < rows.length; i++){
                        var row = rows[i];

                        var j = 0;
                        fields.forEach(function(f){
                            if(f.name != 'latlng' && f.name != 'market' && f.name != 'technology'){
                                j++;

                                q.push({r: row, hk: ('' + row[f.name]).toUpperCase() + '~' + i + '-' + j + '~' + row.latlng}, function(err){
                                    if(err) console.log(err);
                                });
                            }
                        });
                    }
                    console.log('Finished indexing');
                });

                connection.on('error', function(err){
                    return console.log('Database connection error ' + err);
                });
            }
            console.log('Queries processed');
        });
    }
}).listen(port);

dispatcher.onGet('/', function(req, res){
    if(httpStarted){
        res.writeHead(200, {'Content-Type': 'text/html'});
        res.end('MySQL Indexer running' + '\n');

        console.log('MySQL Indexer running');
    }
});

dispatcher.onGet('/status', function(req, res){
    res.writeHead(200, {'Content-Type': 'text/html'});
    res.end('Status ' + q.length() + '\n');

    console.log('Status ' + q.length());
});

一些补充是添加一个 kill/link 来终止进程并改进状态报告，仅举几例。