Golang中使用 Mqtt

MQTT 是一种基于发布/订阅模式的 轻量级物联网消息传输协议 ，可以用极少的代码和带宽为联网设备提供实时可靠的消息服务，它广泛应用于物联网、移动互联网、智能硬件、车联网、电力能源等行业。

本文主要介绍如何在 Golang 项目中使用 github.com/eclipse/paho.mqtt.golang 客户端库 ，实现客户端与 MQTT 服务器 的连接、订阅、收发消息等功能。

项目初始化

环境为1.23.2

本项目使用 paho.mqtt.golang 作为 MQTT 客户端库，安装：

go get github.com/eclipse/paho.mqtt.golang

连接Mqtt

	opts := mqtt.NewClientOptions().AddBroker("tcp://broker.emqx.io:1883")
	opts.SetClientID("mqtt_golang_NTkxOD123213")  // Client ID
	// opts.SetUsername("mqtt_toys")  // 用户名
	// opts.SetPassword("to113gz")  // 用户密码
	opts.SetDefaultPublishHandler(onMessageReceived)  // 订阅主题时的消息处理函数

	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		log.Fatal(token.Error())
		os.Exit(1)
	}

	// 订阅主题
	// production/# 匹配 production/ 开头的主题 
	if token := client.Subscribe("production/#", 0, nil); token.Wait() && token.Error() != nil {
		log.Fatal(token.Error())
		os.Exit(1)
	}

订阅主题消息处理函数

func onMessageReceived(client mqtt.Client, message mqtt.Message) {
	now := time.Now()
	fmt.Printf("时间：%s\t接收topic: %s\tMessage: %s\n", now.Format("2006-01-02 15:04:05.000"), message.Topic(), message.Payload())
	// 在这里将消息转发回业务平台，您可以根据需要修改此部分
 
}

发送主题

	// 玩具入库数据
	toysProduce := map[string]interface{}{
		"method": "produce",
		"params": map[string]interface{}{
			"sex": "1", 
			"name": "test",
			"ver": "V1.0.0",
		},
	}

	mjson, _ := json.Marshal(toysProduce) //转json 
	// 发送代码指令
	token := client.Publish("production/create", 0, false, string(mjson))
	token.Wait()

完成代码

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

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

func onMessageReceived(client mqtt.Client, message mqtt.Message) {
	now := time.Now()
	fmt.Printf("时间：%s\t接收topic: %s\tMessage: %s\n", now.Format("2006-01-02 15:04:05.000"), message.Topic(), message.Payload())
	// 在这里将消息转发回业务平台，您可以根据需要修改此部分 
}

func main() {
opts := mqtt.NewClientOptions().AddBroker("tcp://broker.emqx.io:1883")
	opts.SetClientID("mqtt_golang_NTkxOD123213")  // Client ID
	// opts.SetUsername("mqtt_toys")  // 用户名
	// opts.SetPassword("to113gz")  // 用户密码
	opts.SetDefaultPublishHandler(onMessageReceived)  // 订阅主题时的消息处理函数

	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		log.Fatal(token.Error())
		os.Exit(1)
	}

	// 订阅主题
	// production/# 匹配 production/ 开头的主题
	if token := client.Subscribe("production/#", 0, nil); token.Wait() && token.Error() != nil {
		log.Fatal(token.Error())
		os.Exit(1)
	}

	// 玩具入库数据
	toysProduce := map[string]interface{}{
		"method": "produce",
		"params": map[string]interface{}{
			"sex": "1", 
			"name": "test",
			"ver": "V1.0.0",
		},
	}

	mjson, _ := json.Marshal(toysProduce) //转json
	fmt.Println("发送数据：", string(mjson))
	// 发送代码指令
	token := client.Publish("production/create", 0, false, string(mjson))
	token.Wait()

	// 处理系统信号，以便在接收到SIGINT或SIGTERM时优雅地关闭程序
	signalChan := make(chan os.Signal, 1)
	signal.Notify(signalChan, os.Interrupt, syscall.SIGTERM)

	<-signalChan
	fmt.Println("Received signal, shutting down...")
	client.Disconnect(250)
}