golang开源的可嵌入应用步伐高性能的MQTT服务

golang开源的可嵌入应用步伐高性能的MQTT服务

什么是MQTT？

MQTT（Message Queuing Telemetry Transport）是一种轻量级的、开放的消息传输协议，计划用于在低带宽、高延长或不可靠的网络环境中举行通讯。MQTT最初由IBM开辟，现已成为OASIS标准。
MQTT的计划目的是提供一种简朴、轻量、可扩展的协议，实用于各种装备和网络条件。它通常用于物联网（IoT）和传感器网络，此中装备须要以有效的方式举行通讯，而且资源（如带宽和电池寿命）大概受到限定。
MQTT的简朴计划和实用性使其成为物联网中常用的通讯协议之一。它被广泛用于传感器网络、嵌入式装备、移动应用步伐和其他场景中，提供了一种可靠、高效的消息传输机制。
什么是Mochi-MQTT

源代码地点：https://github.com/mochi-mqtt/server
Mochi MQTT 是一个完全兼容 MQTT v5的可嵌入的中心件/服务器，完全利用 Go 语言编写，旨在用于遥测和物联网项目的开辟。它可以作为独立的二进制文件利用，也可以嵌入到你本身的应用步伐中库来利用，颠末提出的计划以实现标题的轻量化和快速摆设，同时也非常器重代码的质量和可维护性。
用途

物联网项目开辟时，经常须要利用MQTT协议对装备接入，在很多场景中，私有化摆设物联网体系时资源比力少，性能要求高，一些大型的MQTT服务不满意要求，而且代码不可控。
尚有在边沿场景下，须要在边沿网关，边沿控制器装备上摆设物联网体系，但是边沿网关的资源很少，内存约莫只有4G，以是利用java开辟的物联网体系就很难摆设上去；利用C/C++开辟服从又很低，以是Go语言是最符合的，
Mochi-MQTT刚好又美满是Go编写的开源的，可以嵌入到本身的步伐启动。
Mochi MQTT独立摆设

Golang的环境设置这里不做分析，请看我前面的博文分析
Mochi MQTT 可以作为独立的中心件利用。只需拉取此堆栈代码，然后在 cmd 文件夹中运行 cmd/main.go ，默认将开启下面几个服务端口， tcp (:1883)、websocket (:1882) 和服务状态监控 (:8080) 。

1. cd cmd
2. go build -o mqtt && ./mqtt

复制代码

docker摆设

可以从 Docker Hub 堆栈中拉取并运行Mochi MQTT官方镜像：

1. docker pull mochimqtt/server
2. 或者
3. docker run mochimqtt/server

复制代码

也提供了一个简朴的 Dockerfile，用于运行 cmd/main.go 中的 Websocket(:1882)、TCP(:1883) 和服务端状态信息(:8080)这三个服务监听：

1. docker build -t mochi:latest .
2. docker run -p 1883:1883 -p 1882:1882 -p 8080:8080 mochi:latest

复制代码

嵌入本身项目运行和开辟

下载Mochi MQTT包

1. go get github.com/mochi-mqtt/server/v2

复制代码

将Mochi MQTT作为包导入利用, 示例代码如下

1. import (
2.   mqttServer "github.com/mochi-mqtt/server/v2"
3.         "github.com/mochi-mqtt/server/v2/listeners"
4.         "github.com/mochi-mqtt/server/v2/packets"
5. )
7. var Server *mqttServer.Server
9. func ServerMqttInit() {
10.         // 创建新的 MQTT 服务器。
11.         Server = mqttServer.New(&mqttServer.Options{
12.                 InlineClient: true, // 启动内联客户端
13.         })
14.        
15.         // 初始化数据库实例
16.         edge := &edgeHook{deviceDao: deviceDao.NewDeviceRepository(),
17.                 productDao:     productDao.NewProductRepository(),
18.         }
19.         // 添加自定义权限方法
20.         err := Server.AddHook(edge, nil)
21.         if err != nil {
22.                 log.Fatal(err)
23.         }
25.         // 在1883端口上创建一个 TCP 服务端。
26.         tcp := listeners.NewTCP("t1", ":1883", nil)
27.         err = Server.AddListener(tcp)
28.         if err != nil {
29.                 log.Fatal(err)
30.         }
32.         // 在1882端口上创建一个 Websocket 服务端。
33.         ws := listeners.NewWebsocket("ws1", ":1882", nil)
34.         err = server.AddListener(ws)
35.         if err != nil {
36.                 log.Fatal(err)
37.         }
39.         go func() {
40.                 err := Server.Serve()
41.                 if err != nil {
42.                         log.Fatal(err)
43.                 }
44.         }()
45. }
47. type edgeHook struct {
48.         mqttServer.HookBase
49.         deviceDao      deviceDao.DeviceRepository
50.         productDao     productDao.ProductRepository
51. }
53. func (h *edgeHook) ID() string {
54.         return "mqtt-auth"
55. }
57. func (h *edgeHook) Provides(b byte) bool {
58.         // 实现钩子函数
59.         return bytes.Contains([]byte{
60.                 //MQTT连接时认证. 当用户尝试与服务器进行身份验证时调用。
61.                 mqttServer.OnConnectAuthenticate,
62.                 //MQTT topic权限控制. 当用户尝试发布或订阅主题时调用，用来检测ACL规则。
63.                 mqttServer.OnACLCheck,
64.                 //在新客户端连接并进行身份验证后，会立即调用此方法，并在会话建立和发送CONNACK之前立即调用。
65.                 mqttServer.OnSessionEstablish,
66.                 //当客户端因任何原因断开连接时调用。
67.                 mqttServer.OnDisconnect,
68.                 //当客户端向订阅者发布消息后调用。
69.                 mqttServer.OnPublished,
70.         }, []byte{b})
71. }
73. // OnConnectAuthenticate MQTT连接时认证. 当用户尝试与服务器进行身份验证时调用。
74. func (h *edgeHook) OnConnectAuthenticate(cl *mqttServer.Client, pk packets.Packet) bool {
75.         username := string(pk.Connect.Username)
76.         password := string(pk.Connect.Password)
77.         if username == "" || len(username) == 0 {
78.                 return false
79.         }
80.         if password == "" || len(password) == 0 {
81.                 return false
82.         }
83.         return true
84. }
86. // OnACLCheck MQTT topic权限控制. 当用户尝试发布或订阅主题时调用，用来检测ACL规则。
87. func (h *edgeHook) OnACLCheck(cl *mqttServer.Client, topic string, write bool) bool {
88.         username := string(cl.Properties.Username)
89.         if username == "" || len(username) == 0 {
90.                 return false
91.         }
92.         if topic == "" || len(topic) == 0 {
93.                 return false
94.         }
95.         return true
96. }
98. // OnSessionEstablish 在新客户端连接并进行身份验证后，会立即调用此方法，并在会话建立和发送CONNACK之前立即调用。
99. func (h *edgeHook) OnSessionEstablish(cl *mqttServer.Client, pk packets.Packet) {
100.         username := string(cl.Properties.Username)
101.         if username == "" || len(username) == 0 {
102.                 return
103.         }
104.         //设备连接MQTT成功后保存设备在线状态
105. }
107. // OnDisconnect 当客户端因任何原因断开连接时调用。
108. func (h *edgeHook) OnDisconnect(cl *mqttServer.Client, err error, expire bool) {
109.         username := string(cl.Properties.Username)
110.         if username == "" || len(username) == 0 {
111.                 return
112.         }
113.         //设备断开MQTT成功后保存设备离线状态
114. }
116. // OnPublished 当客户端向订阅者发布消息后调用。
117. func (h *edgeHook) OnPublished(cl *mqttServer.Client, pk packets.Packet) {
118.         Log.Infof("mqtt server OnPublished info topic=%s, msg=%s", pk.TopicName, string(pk.Payload))
119.         //收到客户端消息后做业务逻辑处理
120. }
122. // 使用内联客户端方式，向MQTT发送消息
123. func PublishMsg(topic string, msg []byte) bool {
124.         err := Server.Publish(topic, msg, false, 0)
125.         if err != nil {
126.                 Log.Errorf("mqtt EdgePublish error=%v, topic=%s, msg=%s", err, topic, msg)
127.                 return false
128.         }
129.         return true
130. }
132. // 使用内联客户端方式，订阅边缘MQTT消息topic
133. func SubscribeTopic(topic string, subscriptionId int, callback func(topic string, msg []byte)) {
134.         callbackFn := func(cl *mqttServer.Client, sub packets.Subscription, pk packets.Packet) {
135.                 Log.Info("mqtt EdgeSubscribe received message", "client", cl.ID, "subscriptionId", sub.Identifier,
136.                         "topic", pk.TopicName, "payload", string(pk.Payload))
137.                 callback(pk.TopicName, pk.Payload)
138.         }
139.         _ = Server.Subscribe(topic, subscriptionId, callbackFn)
140. }
142. // 使用内联客户端方式，取消订阅边缘MQTT消息topic
143. func UnsubscribeTopic(topic string, subscriptionId int) {
144.         _ = Server.Unsubscribe(topic, subscriptionId)
145. }
147. func main() {
148.         // 创建信号用于等待服务端关闭信号
149.   sigs := make(chan os.Signal, 1)
150.   done := make(chan bool, 1)
151.   signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
152.   go func() {
153.     <-sigs
154.     done <- true
155.   }()
156.   
157.   <-done
158.         Log.Error("caught signal, stopping...")
159.         Server.Close()
160.         Log.Error("main.go finished")
161. }

复制代码

监控MQTT指标信息

1. mqttRouters := r.Group("/mqtt", func(context *gin.Context) {})
2.         {
3.                 mqttRouters.GET("stats", func(c *gin.Context) {
4.                         util.R(c, nil, mqtt.Server.Info)
5.                 })
6.         }

复制代码

详情利用指南请看：https://github.com/mochi-mqtt/server

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！qidao123.com:ToB企服之家，中国第一个企服评测及软件市场,开放入驻,技术点评得现金