go语言实现mqtt协议的实践

一、什么是mqtt

mqtt（message queuing telemetry transport，消息队列遥测传输协议），是一种基于发布/订阅（publish/subscribe）模式的“轻量级”通讯协议，该协议构建于tcp/ip协议上，由ibm在1999年发布。

mqtt最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。作为一种低开销、低带宽占用的即时通讯协议，使其在物联网、小型设备、移动应用等方面有较广泛的应用

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

mqtt还有一个特点就是客户端之间不用相互通信, mqtt通信更像是邮箱服务，发布者发布消息到服务器，接收者只要订阅了其服务在线后即可收到

实现mqtt协议需要客户端和服务器端通讯完成，在通讯过程中，mqtt协议中有三种身份：发布者（publish）、代理（broker）（服务器）、订阅者（subscribe）。其中，消息的发布者和订阅者都是客户端，消息代理是服务器，消息发布者可以同时是订阅者。

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

（1）topic，可以理解为消息的类型，订阅者订阅（subscribe）后，就会收到该主题的消息内容（payload）；

（2）payload，可以理解为消息的内容，是指订阅者具体要使用的内容。

topic就是消息名，payload就是消息体

mqtt会构建底层网络传输：它将建立客户端到服务器的连接，提供两者之间的一个有序的、无损的、基于字节流的双向传输。

当应用数据通过mqtt网络发送时，mqtt会把与之相关的**服务质量（qos）和主题名（topic）**相关连。

二、go语言mqtt服务器broker的搭建

服务端用erlang编写的一个开源项目：emqqtd

# 下载安装包
wget https://github.com/emqx/emqx/releases/download/v4.0.4/emqx-ubuntu18.04-v4.0.4.zip
cd mqttd/emqx
.
├── bin
├── data
├── erts-10.5.2
├── etc
├── lib
├── log
└── releases
# 开启服务
./bin/emqx start
# 查看状态
./bin/emqx_ctl status
# 停止服务
./bin/emqx stop

找到自己的ip，访问http://[你的ip]:18083/#/clients

• 用户名：admin
• 密码：public

即可进入服务器的控制台

三、go客户端访问简单api

客户端用golang客户端的库：“github.com/eclipse/paho.mqtt.golang”

# 下载依赖包
go get -u github.com/eclipse/paho.mqtt.golang

实例如下：

编写了两个函数一个发布一个订阅，传入参数即可服务

修改emqserveraddress为你服务器的ip

package main

// 与后端mqtt服务交互

import (
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"log"
	"os"
	"strconv"
	"time"
)

const emqserveraddress = "你的ip"

// 创建全局mqtt publish消息处理 handler
var messagepubhandler mqtt.messagehandler = func(client mqtt.client, msg mqtt.message) {
	fmt.println("push message:")
	fmt.printf("topic: %s\n", msg.topic())
	fmt.printf("msg: %s\n", msg.payload())
}

// 创建全局mqtt sub消息处理 handler
var messagesubhandler mqtt.messagehandler = func(client mqtt.client, msg mqtt.message) {
	fmt.println("收到订阅消息:")
	fmt.printf("sub client topic : %s \n", msg.topic())
	fmt.printf("sub client msg : %s \n", msg.payload())
}

// 连接的回掉函数
var connecthandler mqtt.onconnecthandler =func(client mqtt.client) {
	fmt.println("新的连接!" + " connected")
}

// 丢失连接的回掉函数
var connectlosthandler mqtt.connectionlosthandler = func(client mqtt.client, err error) {
	fmt.printf("connect loss: %v\n", err)
}

func init() {
	// 配置错误提示
	mqtt.debug = log.new(os.stdout, "		[mqttdebug]", 0)
	mqtt.error = log.new(os.stdout, " 	[mqtterror]", 0)
}

/**
 * @description: 发布订阅
 * @param clientid
 * @param addr
 * @param topic
 * @param payload
 */
func push(topic string, qos byte, retain bool, payload string) {
	// opts clientoptions 用于设置 broker，端口，客户端 id ，用户名密码等选项
	opts := mqtt.newclientoptions().addbroker("tcp://" + emqserveraddress + ":1883").setclientid("test_push")
	opts.setkeepalive(60 * time.second)
	// message callback handler，在没有任何订阅时，发布端调用此函数
	opts.setdefaultpublishhandler(messagepubhandler)
	opts.setpingtimeout(1 * time.second)
	opts.onconnect = connecthandler
	opts.onconnectionlost = connectlosthandler
	client := mqtt.newclient(opts)
	if token := client.connect(); token.wait() && token.error() != nil {
		panic(token.error())
	}
	//发布消息
	// qos是服务质量: ==1: 一次, >=1: 至少一次, <=1:最多一次
	// retained: 表示mqtt服务器要保留这次推送的信息，如果有新的订阅者出现，就会把这消息推送给它（持久化推送）
	token := client.publish(topic, qos, retain, payload)
	token.wait()
	fmt.println("push data : "+topic, "data size is "+strconv.itoa(len(payload)))
	fmt.println("disconnect with broker")
	client.disconnect(250)
}

/**
 * @description: 订阅与取消订阅
 * @param clientid
 * @param addr
 * @param topic
 * @param issub
 */
func subscription(topic string, qos byte, issub bool, handlefun func([]byte)) {
	opts := mqtt.newclientoptions().addbroker("tcp://" + emqserveraddress + ":1883").setclientid("sub_test")
	opts.setkeepalive(60 * time.second)
	opts.setpingtimeout(1 * time.second)
	opts.onconnect = func(client mqtt.client) {
		fmt.println("new subscription! connected" + " => " + topic)
	}
	opts.onconnectionlost = connectlosthandler
	client := mqtt.newclient(opts)
	if token := client.connect(); token.wait() && token.error() != nil {
		panic(token.error())
	}

	if issub {
		// 订阅消息
		if token := client.subscribe(topic, qos, func(client mqtt.client, msg mqtt.message) {
			fmt.printf("receive subscribe message :")
			fmt.printf("sub client topic : %s, data size is  %d \n", msg.topic(), len(msg.payload()))
			if len(msg.payload()) > 0 {
				handlefun(msg.payload())
			}
		}); token.wait() && token.error() != nil {
			fmt.println(token.error())
			os.exit(1)
		}
	} else {
		// 取消订阅
		if token := client.unsubscribe(topic); token.wait() && token.error() != nil {
			fmt.println(token.error())
			os.exit(1)
		}
	}
}

学习资料：

到此这篇关于go语言实现mqtt协议的实践的文章就介绍到这了,更多相关go语言 mqtt协议内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！