Nodejs 冻结了 MySQL 的大量结果输入 REDIS HMSET答案

【问题标题】：Nodejs freezes with MySQL large set of results feed into REDIS HMSETNodejs 冻结了 MySQL 的大量结果输入 REDIS HMSET
【发布时间】：2015-09-18 22:27:00
【问题描述】：

我有一个脚本，它使用 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);

谁能指出我有什么问题或如何改进大型数据集？谢谢！

在添加用于使用队列的异步库的指针之后，我修改了代码，但得到了一个令人讨厌的错误：

{ [错误：连接丢失：服务器关闭连接。] 致命：true，代码：'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 池机制，但我仍然有断开连接，请帮助。另外，实际插入到 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);

【问题讨论】：

如果只执行第二个查询会发生什么？另外，我假设第二个查询根本不依赖于第一个查询首先完成，对吗？因为您是同时运行查询，而不是连续运行。
根据我从 console.log 得到的反馈，我只得到 1,059,952（其中要存储的数据是 529,976 x 4 = 2,119,904），所以我大约在中间，然后似乎什么都没有发生。
Redis 很可能已停止，因为您正在从单个客户端发出近 226,393*4 + 529,976 * 4 个并发请求。
谢谢大家，请指点我如何解决这个问题？
看看这个关于 Redis 断开连接。 exratione.com/2013/01/… 不确定这是不是答案，但值得知道。

标签： javascript mysql node.js redis

【解决方案1】：

感谢大家帮助我，该脚本现在使用 MySQL 池、作业队列和我添加了 HTTP Dispatcher 以获取关于主要任务进度的简单状态报告：将 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());
});

添加一个终止/链接以终止进程并改进状态报告，仅举几例。

【讨论】：