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 mainimport ("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) //转jsonfmt.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)<-signalChanfmt.Println("Received signal, shutting down...")client.Disconnect(250)
}