wordpress最强的教育网站,百度推广产品有哪些,0点开服的网页游戏,网站建设价格差别为什么这么大一、场景描述
昨天在博客《客户端订阅服务端事件的实现方法》中提出了利用websocket、服务端EventEmitter和客户端mitt实现客户端订阅服务端事件#xff0c;大大简化了客户端对服务端数据实时响应的逻辑。上述方案适用于单服务节点的情形。
对于由服务集群支撑的微服务架构大大简化了客户端对服务端数据实时响应的逻辑。上述方案适用于单服务节点的情形。
对于由服务集群支撑的微服务架构websocket提供的点对点通信已无法满足前端订阅后端集群事件的需求升级方案是使用基于消息总线的通信方式。 二、几种消息总线适用性比较
常用的消息总线包括Kafka、Redis和基于MQTT协议实现的EMQX。
Kafka和MQTT都是从发布/订阅系统演化而来但发展侧重点不同。Kafka通过分布式架构提供了海量数据流的存储并保证数据流顺序它的设计目标是支持数据发布、订阅和存储。而MQTT用于网络中传输小型数据包其设计目的是实现简单、可靠的设备间通信。
而Redis是从内存数据库系统演化而来发布/订阅功能是把消息保存在内存中。与Kafka相比其只能提供半持久化与MQTT相比其通信效率较低。
由于应用场景没有对消息持久化的需求且考虑到产业大脑平台未来会接入工业互联网使用MQTT协议来搭建事件总线更利于平台在工业互联网环境下的扩展。
三、MQTT简介
几年前本人曾写过MQTT简介。本文摘抄其中重要概念。
一MQTT协议
MQTTMessage Queuing Telemetry Transport消息队列遥测传输协议是基于“订阅/发布”模式的轻量级通信协议该协议基于TCP/IP能以极低的带宽为海量百万级跨域设备提供可靠的消息服务因此在物联网、小型移动终端、边缘计算方面有广泛应用。 所谓可靠的消息传输体现为可配置消息的服务质量QoS有三种服务质量可选
至多一次 消息发布完全依赖底层TCP/IP网络。会发生消息丢失或重复。应用场景如环境传感器的数据采集丢失一次记录无所谓因为不久后还会有第二次发送。至少一次 确保消息送达订阅者但消息可能重复适用于幂等性操作。只有一次 最严格的消息服务质量确保消息到达且仅到达一次订阅者。应用场景如计费系统等。
MQTT协议中存在三种身份消息总线Broker、发布者Publish和订阅者Subscribe其中消息总线属于服务器后两者都属于客户端。发布者和订阅者可以是各种物联网设备和小型终端消息发布者可以同时也是消息订阅者如下图所示。 MQTT传输的消息分为主题Topic和负载payload两部分
Topic可以理解为消息的类型订阅者订阅Subscribe后就会收到该主题的消息内容payloadpayload可以理解为消息的内容是指订阅者具体要使用的内容。
订阅消息时可以在订阅表达式中使用通配符筛选器对主题进行筛选可同时订阅所匹配的多个主题。 MQTT协议中主要有以下5个方法
connect客户端建立与服务器的连接disconnect等待客户端完成工作后端口与总线的会话subscribe客户端向消息总线注册订阅主题unsubscribe客户端等待消息总线取消所注册的订阅publish客户端向消息总线发送某主题的消息
二开源消息总线EMQX
EMQXErlang/Enterprise/Elastic MQTT Broker是基于Erlang语言开发的开源物联网MQTT消息总线。其是一款由前华为员工开发的开源软件软件主页为https://www.emqx.io/。可根据操作系统类别选择不同版本下载安装或通过docker部署。
软件安装后通过 emqx start以后台方式启动。启动后将会开放两个端口
18083端口为控制台端口可通过浏览器访问该端口首次登录的用户名和密码为admin和public。控制台提供了总线监控、用户权限管理、在线客户端订阅/发布等功能。8083端口为通信端口MQTT客户端可通过该端口与EMQX消息总线通信。
三MQTT.js客户端
MQTT.js是MQTT客户端Nodejs SDK可在浏览器ES模块和Node.js环境CommonJS模块下使用前者可通过MQTT over WebSocket使用后者既可以通过MQTT over WebSocket使用也可以直接使用MQTT。区别仅仅是连接参数的协议头不同。
1. 安装和帮助文件
$ pnpm i mqtt -S #安装
$ npx mqtt help #帮助
MQTT.js command line interface, available commands are:* publish publish a message to the broker* subscribe subscribe for updates from the broker* version the current MQTT.js version* help help about commandsLaunch mqtt help [command] to know more about the commands.2. 使用方法
// const mqtt require(mqtt) //ES模块
import mqtt from mqtt //CommonJS模块// 连接选项
const options {clean: true, // true: 清除会话, false: 保留会话connectTimeout: 4000, // 超时时间// 认证信息clientId: user_id, // 要保证唯一性// 若在控制台配置了用户名和密码// username: xxx,// password: xxx,
}// 连接字符串, 通过协议指定使用的连接方式
// ws 未加密 WebSocket 连接
// wss 加密 WebSocket 连接
// mqtt 未加密 TCP 连接
// mqtts 加密 TCP 连接
// wxs 微信小程序连接
// alis 支付宝小程序连接
const connectUrl ws://localhost:8084/mqtt
const client mqtt.connect(connectUrl, options)client.on(reconnect, error {console.error(正在重连:, error)
})client.on(error, error {console.error(连接失败:, error)
})//收到消息
client.on(message, (topic, message) {console.log(收到消息, topic, message.toString()) //message是二进制流需要转换成字符串
})//订阅主题
const topic/user_id/#
const qos0 //0最多交付1次1至少交付1次2只交付1次
client.subscribe(topic, qos, error{ //订阅user_id主题下所有消息if(error){console.error(订阅主题失败, error)return}console.log(订阅成功)
})//发布消息
client.publish(user_id/a,JSON.stringify({a:123}),qos,error{if(error){console.error(发布消息失败, error)}
})//取消订阅
client.unsubscrib(topic, qos, error{if(error){console.error(取消订阅失败, error)return}
})//断开连接
if(client.connected){try{client.end(false,(){console.log(成功断开连接)})catch(error){console.error(断开连接失败, error)}}
}四安全性
1. 排它订阅
排它订阅是 EMQX 支持的 MQTT 扩展功能。排它订阅允许对主题进行互斥订阅一个主题同一时刻仅被允许存在一个订阅者在当前订阅者未取消订阅前其他订阅者都将无法订阅对应主题。
2. JWT认证
系统整体采用JWT认证方式通过一台认证服务器颁发JWT Token。MQTT客户端访问EMQX总线时携带由认证服务器颁发的JWT Token。
