网页端HTML使用MQTTJs订阅RabbitＭＱ数据

最近在做一个公司的日志组件时有一个问题难住了我。今天问题终于解决了。由于在解决问题中，在网上也查了很多资料都没有一个完整的实例可以参考。所以本着无私分享的目的记录一下完整的解决过程和实例。

需求：做一个统一日志系统可以查看日志列表和一个可以订阅最新日志的页面。通过提供一个封装好日志记录方法的sdk文件将日志统一收集。

通过上面的需求进行我们使用RabbitＭＱ+Mongodb来实现系统。

使用Ｃ#封装一个ＳＤＫ大家都会这里就不说了。Ｃ#连接RabbitＭＱ示例代码也是一堆堆的也没什么好说的。下面重点说一下网页端如何使用JS去订阅RabbitＭＱ收到的最新日志信息。

后端都是使用RabbitＭＱ的ＡＭＱＰ协议，而前端要求在网页HTML上显示数据。我们选择了使用ＭＱＴＴ协议从RabbitＭＱ中订阅数据。

具体步骤：

１、先准备好相关JS库。ＭＱＴＴ有一个叫browserMqtt.js看名字就知道是为浏览器提供的JS库。还有一个封装了操作MQ的JS库　mqfactory.js。最后还要一个jquery.js文件。这样工具就准备好了。JS文件下载

２、ＨＴＭＬ端代码。

<script type="text/javascript" src="~/js/MqJs/jquery.js"></script>
<script type="text/javascript" src="~/js/MqJs/browserMqtt.min.js"></script>
<script type="text/javascript" src="~/js/MqJs/mqfactory.js"></script>
<body><div><lable>Host: </lable><input id="txtHost" placeholder="192.168.1.88" value="10.1.0.7" /><br /><lable>Port: </lable><input id="txtPort" placeholder="15675" value="15675" /><br /><label>UserName: </label><input id="txtUserName" placeholder="username" value="admin" /><br /><label>Password: </label><input id="txtPassword" placeholder="password" value="admin" /><br /><label>Protocol: </label><input id="txtProtocol" placeholder="ws" value="ws" /><br /><input id="btnConnect" type="button" value="Connect RabbitMQ" /></div><div><input id="btnSubscribe" type="button" value="Subscribe" /><input id="btnPublish" type="button" value="Publish" /><br /><input id="btnSSHuanjing" type="button" value="Subscribe Huanjing" /><input id="hdnIsSubscribed" type="hidden" value="" /><input id="btnPubHuanjing" type="button" value="Publish Huanjing"><br />路由:<input id="btnRoutingKey" type="text" value="Dcon/Logs/Client"><br /><input id="txtMessage" type="text" placeholder="Please enter message" /></div><div><label>log:</label><br /><ul id="lstLog"></ul><input id="btnClearLog" type="button" value="Clear Log" /></div>
</body>
<script type="text/javascript">$(function () {var mqclient;//var routingKey = 'Dcon.Logs.ServerWebShow';var message;$('#btnSubscribe').attr('disabled', 'disabled');$('#btnPublish').attr('disabled', 'disabled');$('#btnSSHuanjing').attr('disabled', 'disabled');$('#btnPubHuanjing').attr('disabled', 'disabled');$('#btnConnect').click(function () {var mqttOpts = {host: (() => $('#txtHost').val())(),port: (() => $('#txtPort').val())(),username: (() => $('#txtUserName').val())(),password: (() => $('#txtPassword').val())(),//transformWsUrl方法用于在浏览器中使用MQTT的场景，默认情况下,MQTT自动生成的url为ws://ip:port形式，//然而服务器要求的格式是ws://ip:port/ws，所以MQTT提供了此接口用于在生成url时自定义url格式
                transformWsUrl: (url, opts, client) => { return opts.protocol && opts.protocol == 'ws' ? url + 'ws' : url; },clientId: (() => { return 'mqttjs_' + Math.random().toString(16).substr(2, 8); })()};var biz = {huanjing: function (handler, isOn) {if (isOn !== false) {this.ss(this.topics.huanjing, handler);} else {this.sus(this.topics.huanjing, handler);}},topics: {huanjing: '/hyj/huanjing/monitor'}};//系统初始化时注入连接选项
            mqfactory.inject(mqttOpts, biz);//创建mqclient单例
            mqclient = mqfactory.create();//注册mqclient的连接成功事件
            mqclient.on('connect', mqconnected);});$('#btnSubscribe').click(function () {if ($(this).val() == 'Subscribe') {//订阅成功后，仅注册一次事件(要考虑每次注册事件时，事件处理器调用的次数，如果仅用一次，就用once方法)//routingKey = $("#btnRoutingKey").val();
                mqclient.once('onss', mqSubscribeSuccess);//简单订阅
                mqclient.ss($("#btnRoutingKey").val());} else {mqclient.once('onsus', mqUnsubscribeSuccess)mqclient.sus($("#btnRoutingKey").val());}});$('#btnPublish').click(function () {var msg = $('#txtMessage').val().length > 0 ? $('#txtMessage').val() : guid();if (message === msg) {msg = guid();}message = msg;$('#txtMessage').val(message);//发送消息
            mqclient.pub($("#btnRoutingKey").val(), message);$('#lstLog').append('<li>Send Message: ' + message + '</li>');});$('#btnSSHuanjing').click(function () {if ($(this).val() == 'Subscribe Huanjing') {mqclient.once('onss', mqHJSubscribeSuccess);mqclient.huanjing(onHuanjingMessageArrived);} else {mqclient.once('onsus', mqHJUnsubscribeSuccess);mqclient.huanjing(onHuanjingMessageArrived, false);}});$('#btnPubHuanjing').click(function () {var msg = $('#txtMessage').val().length > 0 ? $('#txtMessage').val() : guid();if (message === msg) {msg = guid();}message = msg;$('#txtMessage').val(message);//发送消息
            mqclient.pub(mqclient.topics.huanjing, message);$('#lstLog').append('<li>Send Huanjing Message: ' + message + '</li>');});$('#btnClearLog').click(function () {$('#lstLog').empty();});function mqconnected() {//alert("mqconnected");
            $('#btnSubscribe').removeAttr('disabled');$('#btnPublish').removeAttr('disabled');$('#btnSSHuanjing').removeAttr('disabled');$('#btnPubHuanjing').removeAttr('disabled');$('#lstLog').append('<li>mqclient connected</li>');}function mqSubscribeSuccess() {//订阅成功，就注册接受消息的方法，此处要接收多次，因此使用了on
            mqclient.on($("#btnRoutingKey").val(), onMessageArrived);$('#btnSubscribe').val('Unsubscribe');$('#lstLog').append('<li>Subscribe successful.' + $("#btnRoutingKey").val()+'</li>');}function mqUnsubscribeSuccess() {//注销订阅，所以将事件处理器解除绑定
            mqclient.off($("#btnRoutingKey").val(), onMessageArrived);$('#btnSubscribe').val('Subscribe');$('#lstLog').append('<li>Unsubscribe successful</li>');}function mqHJSubscribeSuccess() {$('#btnSSHuanjing').val('Unsubscribe Huanjing');$('#lstLog').append('<li>Hanjing Subscribe successful</li>');}function mqHJUnsubscribeSuccess() {$('#btnSSHuanjing').val('Subscribe Huanjing');$('#lstLog').append('<li>Huanjing Unsubscribe successful</li>');}function onMessageArrived(message) {$('#lstLog').append('<li>Receive message: ' + new Date().toString() + '    ' + message.toString() + '</li>');}function onHuanjingMessageArrived(message) {$('#lstLog').append('<li>Receive Huanjing message: ' + new Date().toString() + '    ' + message.toString() + '</li>');}function guid() {function s4() {return Math.floor((1 + Math.random()) * 0x10000).toString(16).substring(1);}return s4() + s4() + '-' + s4() + '-' + s4() + '-' +s4() + '-' + s4() + s4() + s4();}});
</script>

View Code

３.后端代码：

３.１客户端sdk代码

/// <summary>/// 写日志/// </summary>/// <param name="model"></param>public static void Write(LogModel model){//判断写入的日志级别if (model != null && model.LogLevel >= LogLevel){try{var mqMsg = new MqMessage(){MessageBody = JSON.Serialize(model),MessageRouter = SystemConst.RoutingKeyTopic.LogTopic_Producer};//MQHelper.Instance.ProducerMessage_Fanout(mqMsg);
                    MQHelper.Instance.ProducerMessage_Topic(mqMsg);}catch (Exception ex){var errorLog = string.Format("Ip:{0},LogHelper.Write方法异常,{1}", IpHelper.LocalHostIp, ex.Message);//MQHelper.Instance.ProducerMessage_Fanout(new MqMessage() { MessageBody = errorLog });MQHelper.Instance.ProducerMessage_Topic(new MqMessage() { MessageBody = errorLog });}}}

View Code

３.２后端ＭＱ代码：

#region 主题 交换机/// <summary>/// 生产者　客户端调用/// </summary>/// <param name="msg"></param>public void ProducerMessage_Topic(MqMessage msg){try{using (var connection = factory.CreateConnection()){using (var channel = connection.CreateModel()){var body = Encoding.UTF8.GetBytes(msg.MessageBody);channel.BasicPublish(exchange: SystemConst.MqName_LogMq_TopicDefault,routingKey: msg.MessageRouter,basicProperties: null,body: body);Console.WriteLine(" [x] Sent {0}", msg.MessageBody);}}}catch (Exception ex){var exMsg = ex.Message;}}/// <summary>/// 消费者　服务器接收并写入数据库/// 消费方法无法通过参数传入/// EventHandler<BasicDeliverEventArgs> received/// </summary>public void ConsumeMessage_Topic(params string[] routingKeys){if (routingKeys == null || routingKeys.Length == 0){throw new Exception("请指定接收路由");}using (var connection = factory.CreateConnection()){using (var channel = connection.CreateModel()){var queueName = channel.QueueDeclare().QueueName;//获得已经生成的随机队列名//对列与交换机绑定foreach (var rKey in routingKeys){channel.QueueBind(queue: queueName,exchange: SystemConst.MqName_LogMq_TopicDefault,routingKey: rKey);}var consumer = new EventingBasicConsumer(channel);//绑定消费方法consumer.Received += consomer_Received_Topic;//绑定消费者
                    channel.BasicConsume(queue: queueName,autoAck: true,consumer: consumer);Console.WriteLine("日志订阅服务启动成功.");Console.WriteLine(" Press [enter] to exit.");Console.ReadLine();}}}/// <summary>/// 接收通知服务异步的推送/// </summary>/// <param name="sender"></param>/// <param name="e"></param>void consomer_Received_Topic(object sender, BasicDeliverEventArgs e){var body = e.Body;var message = Encoding.UTF8.GetString(body);Console.WriteLine(" [x] {0}", message);
//这里可以增加写入数据库的代码
        }#endregion

View Code

３.３路由

/// <summary>/// 主题路由/// </summary>public class RoutingKeyTopic{/// <summary>/// 生产者/// </summary>public const string LogTopic_Producer = "Dcon.Logs.Client";/// <summary>/// 消息者_日志服务_保存日志/// </summary>public const string LogTopic_Consume_Server_SaveDB = "Dcon.Logs.*";/// <summary>/// 消息者_日志服务_Web显示日志/// </summary>public const string LogTopic_Consume_Server_WebShow = "Dcon.Logs#";//".Logs.Client";/// <summary>/// 消息者_日志服务_Web显示日志/// </summary>public const string LogTopic_Consume_Server_WebShow_T = "*.Logs.Client";//".Logs.Client";/// <summary>/// 消息者_日志服务_ # 接收所有/// </summary>public const string LogTopic_Consume_Server_All = "#";//".Logs.Client";
        }}

View Code

注意点：

１、ＭＱＴＴ的路由是以　/ 来分割的。在RabbitＭＱ中会被转义成　. 如示例中的路由Dcon/Logs/Client会被转换成　Dcon.Logs.Client

２、网页端接收时的路由要和发送端的路由一至。也就是说后端用　Dcon.Logs.Client　来推数据前端就要使用　Dcon/Logs/Client来接收数据。

３、ＭＱＴＴ路由不支持通配符.

４、由于ＭＱＴＴ的ＪＳ库没有提供Topic交换机与路由绑定功能。所以前端接收时不能设置订阅主题交换机名称。如果要和amqp交互只能使用amqp的默认主题交换机名称　amq.topic

运行效果图：

转载于:https://www.cnblogs.com/stevenchen2016/p/7577793.html

网页端HTML使用MQTTJs订阅RabbitＭＱ数据相关推荐

单片机数据上传到阿里云物联网平台后，如何在手机端和网页端获取获取数据？
最近遇到了一个需求:单片机上传数据到阿里云物联网平台,手机端或者网页端能够从物联网平台获取数据并显示到出来,再加一个手机端/网页端通过物联网平台控制单片机. 最终采用了阿里云物联网平台提供的方法,使用 ...
RabbitMQ 网页端控制台开启方式
最近使用RabbitMQ发现只有命令行的方式使用RabbitMQ对队列进行管理不够方便,后来发现RabbitMQ提供了网页端控制台的方式!十分的方便!界面截图如下: 具体的安装方法如下: 首先进入ra ...
【网页端GF6-WFV数据下载问题-1】
[网页端GF6-WFV数据下载问题-1] 最近发现一个问题,网页端GF6-WFV数据下载下载的网址又更新了目前正在探索期,但是不难发现,整体的页面还是自己人做的,具体使用情况也越来越人性化.本篇博客 ...
fps测试网页_对话斗鱼云游戏负责人：做强网页端，近3成为新增用户
核心要点: 产品形态:现阶段是网页端,避免跨端的体验问题. 用户规模:高峰时占用几百台服务器,并在持续考虑扩容商业模式:测试期间免费,后端持续接入优质的内容和技术解决方案. 直播平台与云游戏结合的想 ...
python使用pika订阅rabbitmq消息链接被重置问题
最近在做一个运维监控系统的时候,使用python的pika插件订阅rabbitmq消息,程序在运行一段时间后,总是会报链接被充值的错误,具体报错如下: Traceback (most recent c ...
PHP小程序码扫码登录网站,微信扫小程序码实现网页端登录
常见的微信扫码登录有两种这两种方式都需要提交企业资料认证和300元年费,有些想要学习或者自己的网站没有盈利的,其实不舍得花这个钱,特别是个人开发者,没有企业资料去做认证. 既然没法做企业认证,那我们 ...
vlc web 登录账号_怎么在web网页端解绑敬业签绑定的手机号？
支持手机电脑多端同步的便签软件敬业签可以用来记录备忘事项,同时针对记录的备忘事项可设置时间提醒.在使用敬业便签时,为了确保数据的安全,以及在日后使用时方便找回账号密码,可以将便签账户和手机号绑定在一起 ...
应用市场高速下载以及网页端调起APP页面研究与实现
Github博文地址,此处更新可能不是很及时. 好久没写博客了,好大一个坑.正好,最近刚做完应用市场的高速下载功能,便拿来填了这个坑. 话说产品为了增加用户量,提升用户活跃度以及配合推广,更坑爹的是看 ...
美摄云非编系统——网页端实时编辑渲染方案
美摄云非编是一款新型网页端非线性编辑工具,应用WebAssembly技术实现网页端直接渲染图像.本次LiveVideoStackCon 2020线上峰会我们邀请到了北京美摄网络科技有限公司的研发总监黄 ...

网页端HTML使用MQTTJs订阅RabbitＭＱ数据

网页端HTML使用MQTTJs订阅RabbitＭＱ数据相关推荐

最新文章

热门文章