MQTT

新的项目需求,使用到了硬件链接,运用到了mqtt技术,查询了相关资料,总结了一下

1、MQTT简介

　　MQTT（Message Queuing Telemetry Transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的轻量级协议，该协议构建于TCP/IP协议之上，MQTT最大优势在于，能够以极少的代码和有限的带宽，为链接远程设备提供实时可靠的消息服务。做为一种低开销、低带宽占用的即时通信协议，使其在物联网、小型设备、移动应用等方面有较普遍的应用。

　　MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特色使它适用范围很是普遍。在不少状况下，包括受限的环境中，如：机器与机器（M2M）通讯和物联网（IoT）。其在，经过卫星链路通讯传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已普遍使用。　　

2、特性

　　MQTT协议工做在低带宽、不可靠的网络的远程传感器和控制设备通信而设计的协议，它具备如下主要的几项特性：

　　（1）使用发布/订阅消息模式，提供一对多的消息发布，解除应用程序耦合。

　　（2）对负载内容屏蔽的消息传输。

　　（3）使用TCP/IP提供网络链接。

　　主流的MQTT是基于TCP链接进行数据推送的，可是一样有基于UDP的版本，叫作MQTT-SN。这两种版本因为基于不一样的链接方式，优缺点天然也就各有不一样了。

　　（4）有三种消息发布服务质量(qss)：

　　“至多一次”，消息发布彻底依赖底层TCP/IP网络。会发生消息丢失或重复。这一级别可用于以下状况，环境传感器数据，丢失一次读记录无所谓，由于不久后还会有第二次发送。这一种方式主要普通APP的推送，假若你的智能设备在消息推送时未联网，推送过去没收到，再次联网也就收不到了。

　　“至少一次”，确保消息到达，但消息重复可能会发生。

　　“只有一次”，确保消息到达一次。在一些要求比较严格的计费系统中，可使用此级别。在计费系统中，消息重复或丢失会致使不正确的结果。这种最高质量的消息发布服务还能够用于即时通信类的APP的推送，确保用户收到且只会收到一次。　

　　

 

 

 固然了这个数字越小,带宽带站越低

3、原理

 

 

实现MQTT协议须要客户端和服务器端通信完成，在通信过程当中，MQTT协议中有三种身份：发布者（Publish）、代理（Broker）（服务器）、订阅者（Subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者能够同时是订阅者。

MQTT传输的消息分为：主题（Topic）和负载（payload）两部分：

• （1）Topic，能够理解为消息的类型，订阅者订阅（Subscribe）后，就会收到该主题的消息内容（payload）；
• （2）payload，能够理解为消息的内容，是指订阅者具体要使用的内容。

4、实现

1.下载emq

 

 

 2.启动后访问

localhost:18083 默认的帐号密码是admin获得下图

 

 

 3.经过node red图形化工具创建链接

 

注意这里的主题就是topic订阅主题,必定要一致

 

 

 

 

 

 正常流程操做完成后能够再emq的图形页面看到客户链接

 

 发送消息后,确实能够收到数据贼表示链接成功

 

 基本上到这里简单mqtt 链接就算实现了

后续不想使用emq,也可使用阿里的 原理其实差很少,只不过有几个id key须要购买,就看我的状况而定了

        public class MqttPublishSample {

        public static void main(String[] args) {

            String topic        = "MQTT Examples";
            String content      = "Message from MqttPublishSample";
            int qos             = 2;
            String broker       = "tcp://mqtt.eclipse.org:1883";
            String clientId     = "JavaSample";
            MemoryPersistence persistence = new MemoryPersistence();

            try {
                MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
                MqttConnectOptions connOpts = new MqttConnectOptions();
                connOpts.setCleanSession(true);
                System.out.println("Connecting to broker: "+broker);
                sampleClient.connect(connOpts);
                System.out.println("Connected");
                System.out.println("Publishing message: "+content);
                MqttMessage message = new MqttMessage(content.getBytes());
                message.setQos(qos);
                sampleClient.publish(topic, message);
                System.out.println("Message published");
                sampleClient.disconnect();
                System.out.println("Disconnected");
                System.exit(0);
            } catch(MqttException me) {
                System.out.println("reason "+me.getReasonCode());
                System.out.println("msg "+me.getMessage());
                System.out.println("loc "+me.getLocalizedMessage());
                System.out.println("cause "+me.getCause());
                System.out.println("excep "+me);
                me.printStackTrace();
            }
        }
    }
记录本周的一个粗心错误->redis读取的服务器地址没有开启,换成本地的就ok了