MQTT概述

MQTT 是用于物联网 （IoT） 的 OASIS 标准消息传递协议。它被设计为一种极其轻量级的发布/订阅消息传递传输，非常适合连接具有小代码占用空间和最小网络带宽的远程设备。 如今，MQTT被广泛应用于各种行业，如汽车、制造、电信、石油和天然气等。

为什么选择MQTT？

轻量化、高效

MQTT客户端非常小，需要最少的资源，因此可以在小型微控制器上使用。MQTT 消息头很小，可以优化网络带宽。

双向通信

MQTT 允许在设备到云以及云到设备之间进行消息传递。这样可以很容易地将消息广播到事物组。

扩展到数百万种事物 

MQTT 可以扩展以连接数百万台物联网设备。

可靠的消息传递

消息传递的可靠性对于许多物联网用例都很重要。这就是为什么 MQTT 有 3 个定义的服务质量级别：0 - 最多一次，1- 至少一次，2 - 正好一次

支持不可靠的网络

许多物联网设备通过不可靠的蜂窝网络进行连接。MQTT 对持久会话的支持减少了将客户端与代理重新连接的时间。

启用安全性

MQTT 使使用 TLS 加密消息和使用现代身份验证协议（如 OAuth）对客户端进行身份验证变得容易。

MQTT 发布/订阅架构

EMQX概述 

EMQ X （简称 EMQ）， 是一款完全开源，高度可伸缩，高可用的分布式 MQTT消息服务器，同时也支持 CoAP/LwM2M 一站式 IoT 协议接入。EMQ 是 5G时代万物互联的消息引擎，适用于 IOT、M2M 和移动应用程序，可处理千万级别的并发客户端。

EMQX部署

本地部署

以ubuntu为例，

wget https://www.emqx.com/zh/downloads/broker/5.3.2/emqx-5.3.2-ubuntu22.04-amd64.deb

sudo dpkg -i emqx-5.3.2-ubuntu22.04-amd64.deb

启动服务

 sudo service emqx start

访问http://localhost:18083/, 默认账号密码admin/public

云服务serverless部署

注册

EMQX Cloud: 全托管的 MQTT 消息云服务

 

建立连接并发送消息

方式一：MQTTX 测试发送消息

下载MQTTX

MQTTX 下载

安装后建立连接

发送消息

方式二：利用代码实现

以PHP为例

安装扩展包

composer require php-mqtt/client

编写代码 

<?php

require('vendor/autoload.php');

use \PhpMqtt\Client\MqttClient;
use \PhpMqtt\Client\ConnectionSettings;

$server   = 'localhost';
$port     = 1883;
$clientId = rand(5, 15);
$username = 'emqx_user';
$password = 'public';
$clean_session = false;
$mqtt_version = MqttClient::MQTT_3_1_1;

$connectionSettings = (new ConnectionSettings)
    ->setUsername($username)
    ->setPassword($password)
    ->setKeepAliveInterval(60)
    ->setLastWillTopic('emqx/test/last-will')
    ->setLastWillMessage('client disconnect')
    ->setLastWillQualityOfService(1);


$mqtt = new MqttClient($server, $port, $clientId, $mqtt_version);

$mqtt->connect($connectionSettings, $clean_session);
printf("client connected\n");

$mqtt->subscribe('emqx/test', function ($topic, $message) {
    printf("Received message on topic [%s]: %s\n", $topic, $message);
}, 0);

for ($i = 0; $i< 10; $i++) {
    $payload = array(
        'protocol' => 'tcp',
        'date' => date('Y-m-d H:i:s'),
        'url' => 'https://github.com/emqx/MQTT-Client-Examples'
    );
    $mqtt->publish(
    // topic
        'emqx/test',
        // payload
        json_encode($payload),
        // qos
        0,
        // retain
        true
    );
    printf("msg $i send\n");
    sleep(1);
}

$mqtt->loop(true);

执行代码

php mqtt.php

以Golang为例

package main

import (
	"fmt"
	"log"
	"os"
	"time"

	"github.com/eclipse/paho.mqtt.golang"package main

import (
	"fmt"
	"log"
	"os"
	"time"

	"github.com/eclipse/paho.mqtt.golang"
)

var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	fmt.Printf("TOPIC: %s\n", msg.Topic())
	fmt.Printf("MSG: %s\n", msg.Payload())
}

func main() {
	mqtt.DEBUG = log.New(os.Stdout, "", 0)
	mqtt.ERROR = log.New(os.Stdout, "", 0)
	opts := mqtt.NewClientOptions().AddBroker("tcp://localhost:1883").SetClientID("emqx_test_client")

	opts.SetKeepAlive(60 * time.Second)
	// 设置消息回调处理函数
	opts.SetDefaultPublishHandler(f)
	opts.SetPingTimeout(1 * time.Second)

	c := mqtt.NewClient(opts)
	if token := c.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	// 订阅主题
	if token := c.Subscribe("testtopic/#", 0, nil); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
		os.Exit(1)
	}

	// 发布消息
	token := c.Publish("testtopic/1", 0, false, "Hello World")
	token.Wait()

	time.Sleep(6 * time.Second)

	// 取消订阅
	if token := c.Unsubscribe("testtopic/#"); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
		os.Exit(1)
	}

	// 断开连接
	c.Disconnect(250)
	time.Sleep(1 * time.Second)
}

)

var f mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	fmt.Printf("TOPIC: %s\n", msg.Topic())
	fmt.Printf("MSG: %s\n", msg.Payload())
}

func main() {
	mqtt.DEBUG = log.New(os.Stdout, "", 0)
	mqtt.ERROR = log.New(os.Stdout, "", 0)
	opts := mqtt.NewClientOptions().AddBroker("tcp://broker.emqx.io:1883").SetClientID("emqx_test_client")
	
	opts.SetKeepAlive(60 * time.Second)
	// 设置消息回调处理函数
	opts.SetDefaultPublishHandler(f)
	opts.SetPingTimeout(1 * time.Second)

	c := mqtt.NewClient(opts)
	if token := c.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	// 订阅主题
	if token := c.Subscribe("testtopic/#", 0, nil); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
		os.Exit(1)
	}
	
	// 发布消息
	token := c.Publish("testtopic/1", 0, false, "Hello World")
	token.Wait()

	time.Sleep(6 * time.Second)

	// 取消订阅
	if token := c.Unsubscribe("testtopic/#"); token.Wait() && token.Error() != nil {
		fmt.Println(token.Error())
		os.Exit(1)
	}
  
  // 断开连接
	c.Disconnect(250)
	time.Sleep(1 * time.Second)
}