MQTT消息服务—

最近的开发中用到MQTT协议和EMQ消息服务器，MQTT和EMQ是物联网的产物，想必会随着5G时代和万物互联的到来而逐渐火热。

# EMQ：
   是一个MQTT消息服务器
   ### 基于 Apache 2.0 协议许可，完全开源。
   EMQ X 的代码都放在 Github 中，用户可以查看所有源代码。
   EMQ X 3.0 支持 MQTT 5.0 协议，是开源社区中第一个支持 5.0 协议规范的消息服务器，并且完全兼容 MQTT V3.1 和 V3.1.1 协议。除了 MQTT 协议之外，EMQ X 还支持别的一些物联网协议（具体请参见下文的 EMQ X Broker 产品功能介绍）。
   ### 单机支持百万连接，集群支持千万级连接；毫秒级消息转发。
   EMQ X 中应用了多种技术以实现上述功能，
   利用 Erlang/OTP 平台的软实时、高并发和容错
   全异步架构
   连接、会话、路由、集群的分层设计
   消息平面和控制平面的分离等
   扩展模块和插件，EMQ X 提供了灵活的扩展机制，可以实现私有协议、认证鉴权、数据持久化、桥接转发和管理控制台等的扩展
   桥接：EMQ X 可以跟别的消息系统进行对接，比如 EMQ X Enterprise 版本中可以支持将消息转发到 Kafka、RabbitMQ 或者别的 EMQ 节点等
   共享订阅：共享订阅支持通过负载均衡的方式在多个订阅者之间来分发 MQTT 消息。比如针对物联网等数据采集场景，会有比较多的设备在发送数据，通过共享订阅的方式可以在订阅端设置多个订阅者来实现这几个订阅者之间的工作负载均衡

  ### MQTT:
   MQTT 是 Message Queuing Telemetry Transport（消息队列遥测传输）的缩写，是 IBM 开发的一个即时通讯协议，它比较适合于在低带宽、不可靠的网络的进行远程传感器和控制设备通讯等，正在日益成为物联网通信协议的重要组成部分。
       基于发布 / 订阅范式的 “轻量级” 消息协议（头部 2 字节）
       专为资源受限的设备、低带宽占用高延时或者不可靠的网络设计，适用于 IoT 与 M2M
       基于 TCP/IP 协议栈
       事实的 IoT 通讯的标准协议
       该协议于 1999 年由 IBM 的 Dr Andy Stanford-Clark 和 Arcom(现为 Eurotech)的 Arlen Nipper 提出，协议版本经历了多次升级和改进，于 2013 年成立 OASIS MQTT 技术规范委员会，并持续发布协议的新版本，
   #### 版本
   2015 年，MQTT3.1.1 协议发布
   2018 年，MQTT5.0 协议发布

#### MQTT 协议的主要特性
       MQTT 协议使用发布 / 订阅消息范式来做到一对多的消息分发以及应用程序的解耦
       MQTT 协议提供了 3 种（QoS）服务质量用于消息传输，适应不同的物联网数据传输场景
       QoS 0：最多一次传送 (只负责传送，发送过后就不管数据的传送情况)
       QoS 1：至少一次传送 (确认数据交付)
       QoS 2：正好一次传送 (保证数据交付成功)
       通过很小的传输开销，以及最小化的协议交换来减少网络流量
       发生异常断线时通知各方的机制

#### MQTT 服务器
       MQTT 服务器是发布者和订阅者之间通信的代理（因此中文也有将 MQTT 服务器翻译为 MQTT 代理），主要提供了以下的功能，
       基于主题的 Pub/Sub 模式，将发布者和订阅着解耦
       对于服务器来说，发布者和订阅者都是“客户端”
       客户端与服务器连接都通过 TCP、TLS 或者 WebSocket
       客户端（发布者）发送一条消息到服务器
       一个或者多个客户端（订阅者）从服务器接收消息

       ##### 按照 MQTT 协议标准，服务器提供三种连接方式，

TCP：默认端口为 1883
TLS：默认端口为 8883
WebSocket：默认端口为 8083

##启动
centos：
emqx start

emqx_ctl status

sudo systemctl start emq

*** sudo service emqx start

windows:
cd emqx/bin
emqx start

java使用
   maven依赖
   <dependency>
       <groupId>org.springframework.integration</groupId>
       <artifactId>spring-integration-mqtt</artifactId>
   </dependency>
   <dependency>
   <groupId>org.eclipse.paho</groupId>
   <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
   <version>1.2.0</version>
   </dependency>

建立连接，
   MemoryPersistence persistence = new MemoryPersistence();
   MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
   //设置连接参数
MqttConnectOptions connOpts = new MqttConnectOptions();
connOpts.setCleanSession(true);

sampleClient.connect(connOpts);

生产者：
向指定的topic发送消息

String content = "Message from MqttPublishSample";
int qos = 2;
MqttMessage message = new MqttMessage(content.getBytes());
message.setQos(qos);

sampleClient.publish(topic, message);

消费者：
从指定的topic订阅消息
String topic = "demo/topics";

sampleClient.subscribe(topic);

sampleClient.setCallback(new MqttCallback() {
public void messageArrived(String topic, MqttMessage message) throws Exception {
String theMsg = MessageFormat.format("{0} is arrived for topic {1}.", new String(message.getPayload()), topic);
System.out.println(theMsg);
}

public void deliveryComplete(IMqttDeliveryToken token) {
}

public void connectionLost(Throwable throwable) {
}
});

集群中，消息生产者通过负载均衡实现主题订阅和消息发布，那消息消费者如何准确的在集群中订阅主题和消费消息
EMQ X 消息服务器每个集群节点，都保存一份主题树(Topic Trie)和路由表。
最终会生成主题树(Topic Trie)和路由表(Route Table):

##集群的配置
   1、修改配置文件etc/emqx.conf
   修改节点名字
   node.name =emqx@172.162.62.97
通过命令

export EMQX_NODE_NAME=emqx@120.74.89.178 && ./bin/emqx start

2、将节点加入集群

./bin/emqx_ctl cluster join emqx@120.78.89.177

创建集群的策略
   manual   手动命令创建集群
   static   静态节点列表自动集群
   mcast   UDP 组播方式自动集群
   dns   DNS A 记录自动集群
   etcd   通过 etcd 自动集群
   k8s   Kubernetes 服务自动集群

配置：
   etc/emqx.conf   EMQ X Edge 消息服务器配置文件
   etc/acl.conf   EMQ X Edge 默认ACL规则配置文件
   etc/plugins/*.conf   EMQ X Edge 各类插件配置文件

用户管理
   1、配置文件emqx.conf
   allow_anonymous = false
   2、添加用户密码
       emqx_ctl plugins load emqx_auth_username #加载用户权限验证插件

emqx_ctl users add <Username> <Password> #设置用户名密码
如：emqx_ctl users add myemq jack666

服务质量qos

MQTT发布消息QoS保证不是端到端的，是客户端与服务器之间的。订阅者收到MQTT消息的QoS级别，最终取决于发布消息的QoS和主题订阅的QoS。只有QoS==1,2时候，才有Store(Msg)之说,，通过Store(Msg)和唯一的MessageID来保证消息不会重复接收和发送！；QoS==0时没有

EMQ的共享订阅和保留消息

共享订阅模式：多人订阅，不重复消费，实现消息消费的负载均衡
$queue 和 $share 之间的差异：
$queue 之后的主题中所有消息将轮流发送到客户端，

$share 之后，您可以添加不同的组，例如:

$share/group_1/topic
           $share/group_2/topic
           $share/group_3/topic
           当 EMQ X 向 topic 发送消息时，每个组都会收到该消息，并依次将其发送到该组中的设备。

保留消息，
message.setRetained(true);保留消息每次订阅都会被消费一次，重新发布只会保留最新的一条

MQTT消息服务——EMQ相关推荐

EMQ百万级MQTT消息服务(优化和压测)
如果使用EMQ来承载百万级别的用户连接可以吗?官方的回复是8核心32G的配置能够承载160W台设备的链接,那就究竟性能如何呢?只有你自己试了才知道,本节就带着大家一起有调优系统配置和对EMQ进行压力测 ...
物联网实战-基于开源 MQTT消息服务器EMQ X
物联网协议对比对于物联网,最重要的是在互联网中设备与设备的通讯,现在物联网在internet通信中比较常见的通讯协议包括:HTTP.websocket.XMPP.COAP.MQTT HTTP和web ...
mqtt 负载均衡_EMQ百万级MQTT消息服务(分布式集群)
在强大的单机也比不上集群,EMQ的集群模式很粗暴,只需要把EMQ服务关联在一起然后负载均衡就可以达到集群的效果,这样就算面对1000CK问题也迎刃而解附上: 喵了个咪的博客:w-blog.cnEMQ ...
联合 EMQ 发布云原生物联网消息服务联合解决方案，云上轻松构建 IoT 应用
近日,「DaoCloud 道客」联合「EMQ 映云科技」发布云原生物联网消息服务联合解决方案,并完成「DaoCloud Enterprise云原生应用云平台」和「EMQX Enterprise 企业级 ...
MQTT消息大小流量消耗测试
1. 本文测试了以下内容: 设备连接mqtt消耗流量每次心跳消耗流量设备断开消耗流量订阅topic消耗流量上传消息消耗流量下放消息消耗流量. 2. 使用软件 mqttx(客户端1) emqx ...
千万级车联网 MQTT 消息平台架构设计
在本专题系列文章中,我们将根据 EMQ 在车联网领域的实践经验,从协议选择等理论知识,到平台架构设计等实战操作,与大家分享如何搭建一个可靠.高效.符合行业场景需求的车联网平台. 前言随着整个汽车出行 ...
微众银行的金融级消息服务平台建设实践和思考
来自:阿里巴巴中间件导读: 近年来,随着微服务架构的流行,分布式消息引擎在物联网.分布式事务.实时计算和大规模缓存同步等场景中的应用日益增多.本文将分享微众银行基于RocketMQ构建消息服务平台的 ...
EMQ 携手 NNG 联合发布新一代超轻量边缘 MQTT 消息引擎 NanoMQ
2021 年伊始,物联网开源基础软件领导者 EMQ 携手 NNG 为开源社区献上开年贺礼:面向边缘计算和 5G MEC 的开源轻量级边缘 MQTT 消息引擎--NanoMQ(https://nanom ...
emq的客户端与服务端_使用 EMQ X Cloud 物联网 MQTT 云服务
使用 EMQ X Cloud 物联网 MQTT 云服务 2020-09-01 摘要在数分钟内创建全托管高可用 MQTT 集群,快速接入物联网设备并立即开始产品原型设计与应用开发,将物联网数据存储到华 ...

MQTT消息服务——EMQ

MQTT消息服务——EMQ相关推荐

最新文章

热门文章