Express + Socket.IO + RabbitMQ (node-amqp)

Question

我很难将这三个内容整合在一起，可能是因为我没有正确理解使用Express进行路由的概念。

我有一个RabbitMQ队列，其中包含事件更新。我们可以通过其ID识别这些事件。因此，我想在给定页面上获取与其ID对应的更新。

队列：1316、1539、3486、3479、1316、3890、3479... ->无限期地从数据库中提取。 www.example.com/event/1316 ->仅获取ID为1316的消息 www.example.com/event/3479 ->仅获取ID为3479的消息

我的代码在加载第一个事件时运行良好，但当我在不同的窗口中加载第二个事件时，它会从两个事件中获取消息，如果我加载第三个事件，那么它就会从这三个ID中获取消息。 app.js

var express = require('express')  
, http = require('http');
var app = express();
var server = http.createServer(app);
var io = require('socket.io').listen(server, { log: false });
require('./io')(io);

var amqp = require('amqp');
var rabbitMQ = amqp.createConnection({ host: 'localhost' });

rabbitMQ.on('ready', function() {
  console.log('Connected to RabbitMQ');
  io.sockets.on('connection', function (socket) {
    console.log('Socket connected: ' + socket.id);
    rabbitMQ.queue('offer', { autoDelete: false, durable: false, exclusive: false }, function(q) {    
      q.bind('#'); // Catch all messages    
      q.subscribe(function (message) {
        obj = JSON.parse(message.data.toString());
        //socket.broadcast.to(obj.id).emit('message', obj);
        io.sockets.in(obj.id).emit('message', obj);
      });
    });
  });
});

var routes = require('./routes')
, event = require('./routes/event');

app.get('/', routes.index);
app.get('/event/:id', event.index);

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

io.js

var socketio = function (io) { 
  if (!io) return socketio._io;  
  socketio._io = io;
} 

module.exports = socketio;

routes/event.js

var io = require('../io')();

exports.index = function(req, res) {
  io.sockets.on('connection', function (socket) {
    socket.join(req.params.id);
  });
  res.render('event', { title: 'Event' });
};

谢谢！

- Puigcerber

Answer 1

在尝试并失败之后，我明白了我在路由器内使用io.sockets.on('connection')导致事件重复的原因。所以最后，最简单的思路是正确的。

app.js

var room = '';
var roomHandler = function(req, res, next) {
  if (req.path.match('event')) {
    room = req.params.id;
  } 
  next(); // Passing the request to the next handler in the stack.
}

io.sockets.on('connection', function (socket) {    
  socket.join(room);
});

rabbitMQ.on('ready', function() { 
  rabbitMQ.queue('offer', { autoDelete: false, durable: false, exclusive: false }, function(q) {      
    q.bind('#'); // Catch all messages  
    q.subscribe(function (message) {
      obj = JSON.parse(message.data.toString());
      io.sockets.in(obj.id).emit('message', obj);
    });
  });
});

app.get('/', routes.index);
app.get('/event/:id', roomHandler, event.index);

- Puigcerber

Answer 2

您之所以收到所有消息，是因为您加入房间后从未离开。如果您查看维基上的Socket IO Rooms，在底部，它提供了io.sockets.manager.roomClients[socket.id]作为一种获取套接字已加入的房间列表的方法（如果您访问了所有三个链接，它们将全部包括在内）。

您可能希望尝试浏览此房间列表并离开任何不是当前房间的房间，看看是否解决了问题。

编辑

好的，所以，有两个原因/解决方案。我刚刚测试了我的理论，它是有效的-您会收到您加入的每个房间的消息，并将继续这样做，直到您将其离开。因此，以下是可选项：

1.在加入房间时离开所有其他房间

io.sockets.on('connection', function (socket) {
    var room = req.params.id;

    var roomKeys = Object.keys(io.sockets.manager.roomClients[socket.id]);
    roomKeys.forEach(function(key) {
        if (key === '' || key === '/' + room) return;
        socket.leave(key.slice(1));
    });

    socket.join(room);
});

正如所说，我已经测试过了，它可以工作。

2. 不要发送 message 事件，发送一个 {room name} 事件

你可以发送一个 '{room name}' 事件，而不是发射一个 'message' 事件。你只需使用 socket.emit(obj.id, obj); 替换 q.subscribe() 回调中的 io.sockets.in(obj.id).emit('message', obj); ，并且你只需要让javascript客户端监听该页面的事件类型（基于URL路径）。

我也测试过了，它同样有效。这种方法更简单（我认为），因为它仅需要在你的 q.subscribe() 回调中使用 .emit()，这意味着你可以省略 'room management' 相关的操作。

- floatingLomas

---

但是我不想离开房间。我的意思是，对于每个事件页面，我希望有一个带有其ID的房间，在那里我可以广播包含该ID的所有消息。 - Puigcerber

我相信你的连接客户端在加入他们访问的每个房间时，却没有在离开页面时退出，因此前往/event/1234然后前往/event/2345意味着他们加入了这两个房间，因此他们会在浏览器中同时接收到两个源 - 这就是你所看到的，对吗？ - floatingLomas

现在我更加迷茫了。假设我想在五个不同的窗口中有五个事件。据我所知，每个事件都会创建一个新的套接字。我想要的是让这些套接字中的每一个都有一个不同的房间（或其他解决方案），因为在事件页面上，我只想获取与其ID对应的事件更新。所以我不明白为什么当我加载第二个窗口时，这个套接字也订阅了第一个ID。 - Puigcerber

所有的/event/nnnn页面都使用同一个套接字。当您从一个页面切换到另一个页面时，仍然在同一个套接字上，因此仍然连接到先前使用该套接字加入的任何房间。因此，当您调用.join()加入新房间时，它会添加新房间 - 但您仍然是以前房间的成员，因此仍然会收到那些消息。因为它是同一个套接字，离开一个页面并不意味着离开一个房间。这样清楚了吗？ - floatingLomas

但是我将每个事件加载到不同的窗口上，同时打开几个窗口，我可以在服务器上看到这些套接字具有不同的ID。 - Puigcerber