node.js - Socket.io、Node.js 和 rethinkdb

Question

我在 Socket.io、Node.js 和 rethinkdb 中使用简单的应用程序。在服务器中，我的 server.js 文件工作了大约 20 分钟。大约 20 分钟后，我的代码停止了。我的服务器 .js :

var express = require('express'),
path = require('path'),
http = require('http'),
io = require('socket.io'),
r = require('rethinkdb');


var app = express();

app.configure(function() {
app.set('port', process.env.PORT || 3000);
app.use(express.logger('dev'));
app.use(express.bodyParser())
app.use(express.static(path.join(__dirname, 'public')));
});

var server = http.createServer(app);
io = io.listen(server);


server.listen(app.get('port'), function() {
console.log("Express server listening on port " + app.get('port'));
});

var count = 0;
var say = 0;
io.sockets.on('connection', function(socket) {

count++;
socket.on('message', function(message) {
    url = message;
    //socket.join(url);
    r.connect({
        host: 'localhost',
        port: 28015
    }, function(err, conn) {
        if (err) throw err;
        r.db('test').table('online').insert({
            socket_id: socket.id,
            news_id: message
        }).run(conn, function(err) {
            if (err) throw err;
            r.db('test').table("online").filter({
                news_id: message
            }).count().run(conn, function(err, res) {
                if (err) throw err;
                say = res;


                console.log("connect say" + say);
                // ip = socket.handshake.address.address;              
                io.sockets.emit('pageview', {
                    'say': say,
                    'count': count,
                    'news_id': message
                });

            });

        });




    });


});



socket.on('disconnect', function() {

    var news_id = 0;
    console.log("Socket disconnected:" + socket.id);

    r.connect({
        host: 'localhost',
        port: 28015
    }, function(err, conn) {

        if (err) throw err;

        r.db('test').table("online").filter({
            socket_id: socket.id
        })("news_id").run(conn).then(function(cursor) {


            cursor.each(function(err, item) {

                news_id = item;


                r.db('test').table("online").filter({
                    socket_id: socket.id
                }).delete().run(conn, function() {

                    r.db('test').table("online").filter({
                        news_id: news_id
                    }).count().run(conn, function(err, res) {
                        if (err) throw err;
                        say = res;
                        count--;
                        console.log("disconnect say" + say);
                        io.sockets.emit('pageview', {
                            'say': say,
                            'count': count,
                            'news_id': news_id
                        });

                    });

                });


                //console.log("news_id"+news_id);




            });
        });




    });




});

});

我的客户端.js:

 var socket = io.connect("http://192.168.1.198:3000");


    socket.on('connect', function () {

        console.log('Socket connected');
        socket.send(news_id);
        socket.on('pageview', function (msg) {


            if(msg.news_id == news_id)
                        {
                                //$('.count_icon').html(msg.say);
                                console.log(msg.say);

                        }


        });

    });

当 server.js 运行时我得到这个错误:

Unhandled rejection RqlDriverError: First argument to `run` must be an open connection.
at new RqlDriverError (/root/node_modules/rethinkdb/errors.js:14:13)
at Bracket.TermBase.run (/root/node_modules/rethinkdb/ast.js:129:29)
at /home/node1/server.js:82:90
at tryCatcher (/root/node_modules/rethinkdb/node_modules/bluebird/js/main/util.js:24:31)
at Promise.errorAdapter (/root/node_modules/rethinkdb/node_modules/bluebird/js/main/nodeify.js:35:34)
at Promise._settlePromiseAt (/root/node_modules/rethinkdb/node_modules/bluebird/js/main/promise.js:528:21)
at Promise._settlePromises (/root/node_modules/rethinkdb/node_modules/bluebird/js/main/promise.js:646:14)
at Async._drainQueue (/root/node_modules/rethinkdb/node_modules/bluebird/js/main/async.js:177:16)
at Async._drainQueues (/root/node_modules/rethinkdb/node_modules/bluebird/js/main/async.js:187:10)
at Async.drainQueues (/root/node_modules/rethinkdb/node_modules/bluebird/js/main/async.js:15:14)
at process._tickCallback (node.js:442:13)

Answer 1

问题似乎是您打开了太多连接，而 RethinkDB 可能正在关闭其中一些连接。您可以添加逻辑以在每次打开这些连接时关闭它们，您可以使用一个全局连接，也可以使用类似 rethinkdbdash 的东西。它通过使用连接池为您处理。

这是使用全局连接(并使用 promises)时您的代码的样子。

var express = require('express'),
path = require('path'),
http = require('http'),
io = require('socket.io'),
r = require('rethinkdb');


var app = express();

app.configure(function() {
    app.set('port', process.env.PORT || 3000);
    app.use(express.logger('dev'));
    app.use(express.bodyParser())
    app.use(express.static(path.join(__dirname, 'public')));
});

var server = http.createServer(app);
io = io.listen(server);

server.listen(app.get('port'), function() {
    console.log("Express server listening on port " + app.get('port'));
});

var count = 0;
var say = 0;
r.connect({
    host: 'localhost',
    port: 28015
})
.then(function (conn) {
    io.sockets.on('connection', function(socket) {

        count++;
        socket.on('message', function(message) {
            url = message;
            //socket.join(url);
            Promise.resolve()
            .then(function() {
                return r.db('test').table('online').insert({
                    socket_id: socket.id,
                    news_id: message
                }).run(conn)
            })
            .then(function() {
                return r.db('test').table("online").filter({
                    news_id: message
                }).count().run(conn);
            })
            .then(function (res) {
                say = res;
                console.log("connect say" + say);
                io.sockets.emit('pageview', {
                    'say': say,
                    'count': count,
                    'news_id': message
                });
            });
        });

        socket.on('disconnect', function() {
            var news_id = 0;
            console.log("Socket disconnected:" + socket.id);
            Promise.resolve()
                .then(function () {
                    return r.db('test').table("online").filter({
                        socket_id: socket.id
                    })("news_id").run(conn);
                })
                .then(function (cursor) {
                    cursor.each(function(err, item) {
                        news_id = item;
                        Promise.resolve()
                            .then(function () {
                                return r.db('test').table("online").filter({
                                    socket_id: socket.id
                                }).delete().run(conn);
                            })
                            .then(function () {
                                return r.db('test').table("online").filter({
                                    news_id: news_id
                                }).count().run(conn);
                            })
                            .then(function (res) {
                                say = res;
                                count--;
                                console.log("disconnect say" + say);
                                io.sockets.emit('pageview', {
                                    'say': say,
                                    'count': count,
                                    'news_id': news_id
                                });
                            });
                    });
                });
        });
    });
})