- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我正在使用rabbitMQ 和socket.io 编写浏览器通知。除了一种情况外,我的配置工作正常。
当我使用用户登录系统时,它会创建一个通知 UID-userid 队列(目前队列名称是通过查询 oaraeter 发送的,我将尽快实现更复杂的方法解决问题)
如果我在另一个浏览器上使用另一个用户登录,它会创建另一个队列通知-UID-第二个用户ID。
如果我注销其中一个用户,队列就会消失(因为它不持久)。
问题是,当我在另一个 session 上刷新或加载另一个页面时,即使未发送参数队列名称,它也会重新创建第二个队列。
server.js
var amqp = require('amqp');
var app = require('express')();
var http = require('http').Server(app);
var io = require('socket.io')(http);
var rabbitMqConnection = null;
var _queue = null;
var _consumerTag = null;
io.use(function (socket, next) {
var handshakeData = socket.handshake;
// Here i will implement token verification
console.log(socket.handshake.query.queueName);
next();
});
// Gets the connection event form client
io.sockets.on('connection', function (socket) {
var queueName = socket.handshake.query.queueName;
console.log("Socket Connected");
// Connects to rabbiMq
rabbitMqConnection = amqp.createConnection({host: 'localhost', reconnect: false});
// Update our stored tag when it changes
rabbitMqConnection.on('tag.change', function (event) {
if (_consumerTag === event.oldConsumerTag) {
_consumerTag = event.consumerTag;
// Consider unsubscribing from the old tag just in case it lingers
_queue.unsubscribe(event.oldConsumerTag);
}
});
// Listen for ready event
rabbitMqConnection.on('ready', function () {
console.log('Connected to rabbitMQ');
// Listen to the queue
rabbitMqConnection.queue(queueName, {
closeChannelOnUnsubscribe: true,
durable: false,
autoClose: true
},
function (queue) {
console.log('Connected to ' + queueName);
_queue = queue;
// Bind to the exchange
queue.bind('users.direct', queueName);
queue.subscribe({ack: false, prefetchCount: 1}, function (message, headers, deliveryInfo, ack) {
console.log("Received a message from route " + deliveryInfo.routingKey);
socket.emit('notification', message);
//ack.acknowledge();
}).addCallback(function (res) {
// Hold on to the consumer tag so we can unsubscribe later
_consumerTag = res.consumerTag;
});
});
});
// Listen for disconnection
socket.on('disconnect', function () {
_queue.unsubscribe(_consumerTag);
rabbitMqConnection.disconnect();
console.log("Socket Disconnected");
});
});
http.listen(8080);
client.js
var io = require('socket.io-client');
$(document).ready(function () {
var socket = io('http://myserver.it:8080/', {
query: { queueName: 'notification-UID-' + UID},
'sync disconnect on unload': true,
});
socket.on('notification', function (data) {
console.log(data);
});
})
有什么想法吗?
最佳答案
所以我解决了我的问题,这是一个变量范围问题,把事情搞乱了。让我解释一下我在做什么,也许这对某人有用。
基本上,我正在尝试创建一个浏览器通知系统,这意味着我的应用程序发布(生产者端)到交换通知对象,其中包含一些信息,例如主题、链接和消息。
交换器是一个fanout (users.notification.fanout),它有两个绑定(bind)交换器:users.direct(直接类型)和users.notification.store(扇出类型)。
当生产者发布通知时,它会使用路由键“notification-UID-userid”(其中 userid 是真实用户 ID)发送到 users.notification.fanout。
通知对象同时访问 users.direct 和 users.notification.store最后一个有一个消费者,在用户未登录的情况下将通知写入数据库,第一个将通知发布到浏览器。
那么浏览器消费者是如何工作的呢?
我使用了 socket.io、节点服务器和 amqplib 的经典组合。
每次用户登录时,socket.io 都会创建一个带有名称和路由键 notification-UID-userid 的队列,并将其绑定(bind)到 users.direct Exchange。
与此同时,我已将 https 添加到我的服务器,因此情况与第一个版本相比发生了一些变化。
您可以阅读评论以了解它的作用。
所以我的server.js是
var amqp = require('amqp');
var fs = require('fs');
var app = require('express')();
// Https server, certificates and private key added
var https = require('https').Server({
key: fs.readFileSync('/home/www/site/privkey.pem'),
cert: fs.readFileSync('/home/www/site/fullchain.pem')},app);
var io = require('socket.io')(https);
// Used to verify if token is valid
// If not it will discard connection
io.use(function (socket, next) {
var handshakeData = socket.handshake;
// Here i will implement token verification
console.log("Check this token: " + handshakeData.query.token);
next();
});
// Gets the connection event from client
io.sockets.on('connection', function (socket) {
// Connection log
console.log("Socket Connected with ID: " + socket.id);
// THIS WAS THE PROBLEM
// Local variables for connections
// Former i've put these variables outside the connection so at
// every client they were "overridden".
// RabbitMq Connection (Just for current client)
var _rabbitMqConnection = null;
// Queue (just for current client)
var _queue = null;
// Consumer tag (just for current client)
var _consumerTag = null;
// Queue name and routing key for current user
var queueName = socket.handshake.query.queueName;
// Connects to rabbiMq with default data to localhost guest guest
_rabbitMqConnection = amqp.createConnection();
// Connection ready
_rabbitMqConnection.on('ready', function () {
// Connection log
console.log('#' + socket.id + ' - Connected to RabbitMQ');
// Creates the queue (default is transient and autodelete)
// https://www.npmjs.com/package/amqp#connectionqueuename-options-opencallback
_rabbitMqConnection.queue(queueName, function (queue) {
// Connection log
console.log('#' + socket.id + ' - Connected to ' + queue.name + ' queue');
// Stores local queue
_queue = queue;
// Bind to the exchange (default)
queue.bind('users.direct', queueName, function () {
// Binding log
console.log('#' + socket.id + ' - Binded to users.direct exchange');
// Consumer definition
queue.subscribe({ack: false}, function (message, headers, deliveryInfo, messageObject) {
// Message log
console.log('#' + socket.id + ' - Received a message from route ' + deliveryInfo.routingKey);
// Emit the message to the client
socket.emit('notification', message);
}).addCallback(function (res) {
// Hold on to the consumer tag so we can unsubscribe later
_consumerTag = res.consumerTag;
// Consumer tag log
console.log('#' + socket.id + ' - Consumer ' + _consumerTag + ' created');
})
});
});
});
// Update our stored tag when it changes
_rabbitMqConnection.on('tag.change', function (event) {
if (_consumerTag === event.oldConsumerTag) {
_consumerTag = event.consumerTag;
// Unsubscribe from the old tag just in case it lingers
_queue.unsubscribe(event.oldConsumerTag);
}
});
// Listen for disconnection
socket.on('disconnect', function () {
_queue.unsubscribe(_consumerTag);
_rabbitMqConnection.disconnect();
console.log('#' + socket.id + ' - Socket Disconnected');
});
});
https.listen(8080);
然后是我的client.js
var io = require('socket.io-client');
$(document).ready(function () {
var socket = io('https://myserver.com:8080/', {
secure: true, // for ssl connections
query: { queueName: 'notification-UID-' + UID, token: JWTToken}, // params sent to server, JWTToken for authentication
'sync disconnect on unload': true // Every time the client unload, socket disconnects
});
socket.on('notification', function (data) {
// Do what you want with your data
console.log(data);
});
})
关于javascript - 即使客户端关闭,Amqp、rabbit mq 和 socket.io 也会重新连接到队列,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51965065/
在 Spring AMQP 项目中, 如果 messageProperties 没有 messageId,它们总是创建 messageId。 像这样.. if (this.createMessageI
我不确定我对 errorHandler 和 returnExceptions 的理解是否正确。 但这是我的目标:我从 App_A 发送了一条消息,使用 @RabbitListener 在 App_B
就目前而言,这个问题不适合我们的问答形式。我们希望答案得到事实、引用资料或专业知识的支持,但这个问题可能会引发辩论、争论、投票或扩展讨论。如果您觉得这个问题可以改进并可能重新打开,visit the
我们希望将大型机放在总线上。我相信是AS400。为此,我想让 CICS 大型机向代理发送 AMQP 消息。有几十个AMQP客户端,包括JMS客户端。我对大型机上可能发生的事情了解不够,无法判断是否可以
我们希望将大型机放在公共(public)汽车上。我相信它是AS400。为此,我想让 CICS 大型机向代理发送 AMQP 消息。有几十个 AMQP 客户端,包括 JMS 客户端。我不太了解大型机上的可
我即将实现一个基于 PHP 的系统,该系统使用 RabbitMQ。我可以看出那里有 2 个成熟的库:PECL AMQP和 php-amqp . 我将同时为客户端和工作人员使用 PHP。 有人对这两个库
我正在使用 RabbitMQ 和 ruby-amqp与 Rails。当 Controller 收到消息时,我执行以下操作: def create AMQP.start("amqp://localh
我有两个困惑。 1.如果消息监听器抛出RuntimeException,SimpleMessageListenrContainer会停止吗?2.如果SimpleMessageListenerConta
我们在绑定(bind)到 Rabbit MQ 实例的 Java 应用程序的日志中遇到以下异常。 这是必须要注意的事情,表示Spring AMQP的实现中存在问题,还是可以忽略的事情?在后一种情况下,此
场景:微服务从 RabbitMQ 队列中获取消息,将其转换为对象,然后微服务对外部服务进行 REST 调用。 它将处理成千上万条这样的消息,如果我们知道外部 Rest 服务已关闭,有没有办法告诉我的消
我有一个使用带有 webflux 的 Boot 2.0 的应用程序,并且有一个端点返回 ServerSentEvent 的 Flux。这些事件是通过利用 spring-amqp 从 RabbitMQ
我正在尝试使用 streadway/amqp 连接到 RabbitMQ 总线Go 的驱动程序。我正在处理重新连接例程,为此,我有一个 rabbitMQConsume 函数调用 rabbitMQConn
我正在尝试通过 SSL 连接到 RabbitMQ。我遵循了此处链接的 RabbitMQ SSL 文档 https://www.rabbitmq.com/ssl.html根据 RabbitMQ SSL
这些 amqp 客户端库之间有什么区别?哪一个是最推荐的?主要区别是什么? 最佳答案 我会推荐 amqp.node和 bramqp通过 node-amqp。 node-amqp 有很多错误并且维护不善
我得到的错误: 2019-12-09 06:39:33.189 ERROR 107132 --- [http-nio-8082-exec-5] o.a.c.c.C.[.[.[/].[dispatche
所以我已经让 MQTT -> MQTT 和 AMQP -> AMQP 工作;不过,MQTT -> AMQP 的翻译似乎在某处不起作用。这是我的测试,如果我的“监听器”也在使用 paho 的 MQTT
我得到的错误: 2019-12-09 06:39:33.189 ERROR 107132 --- [http-nio-8082-exec-5] o.a.c.c.C.[.[.[/].[dispatche
几天前我问了同样的问题:Unable to "Peek" messages from an Azure Service Bus Queue using AMQP and Node 。我再次问同样的问题
我有一个当前与 RabbitMQ 集成的组件。我想将 RabbitMQ 替换为 Azure Event Hub,因为我们现在位于云中。 AMQP 0.9.1 与 AMQP 1.0 兼容吗?交换会无缝进
我有一个当前与 RabbitMQ 集成的组件。我想将 RabbitMQ 替换为 Azure Event Hub,因为我们现在位于云中。 AMQP 0.9.1 与 AMQP 1.0 兼容吗?交换会无缝进
我是一名优秀的程序员,十分优秀!