物联网设备有望改变我们与周围世界互动的方式。支持物联网的设备广泛应用于汽车、农业和许多其他行业。从所有这些设备检索数据不需要付出巨大的努力,这也不需要 mqtt。
mqtt 是物联网消息传递的标准。它的开发始于 90 年代末,并且有一个已被许多人采用的正式规范。 mqtt 提供了一种简单、轻量级的发布/订阅消息协议,使其易于在设备上实现并易于集成到大型系统中。
mqtt 通过使用代理来工作。代理位于物联网设备和使用设备数据的任何应用程序之间。在 mqtt 世界中,设备和应用程序都被视为“客户端”。
如果设备需要向世界发送数据,它就像 mqtt 客户端一样,创建与代理的连接,并将其数据发布到给定主题。 mqtt 主题是代理如何路由数据的方式。当应用程序想要从给定设备接收数据时,它会创建与代理的连接,然后订阅给定设备发布的主题。
为了进行演示,我将展示一个用于发布到 mqtt 代理的简单 go 应用程序。这只是模拟从物联网设备发送消息。 _(请注意,此 go 应用程序将在支持 golang 的嵌入式设备(例如 raspberry pi)上正常运行)_
要运行此示例,您需要查看我之前的文章,基本的 mqtt docker 部署,其中包括一个用于运行 mqtt 代理的 dockerfile。
package main import ( "fmt" "math/rand" "time" mqtt "github.com/eclipse/paho.mqtt.golang" ) const ( broker = "tcp://localhost:1883" clientid = "go-mqtt-client" topic = "iot-messages" ) var connecthandler mqtt.onconnecthandler = func(client mqtt.client) { fmt.println("connected to mqtt broker") } var connectlosthandler mqtt.connectionlosthandler = func(client mqtt.client, err error) { fmt.printf("connection lost: %v", err) } func main() { opts := mqtt.newclientoptions() opts.addbroker(broker) opts.setclientid(clientid) opts.onconnect = connecthandler opts.onconnectionlost = connectlosthandler client := mqtt.newclient(opts) if token := client.connect(); token.wait() && token.error() != nil { panic(token.error()) } for { message := generaterandommessage() token := client.publish(topic, 0, false, message) token.wait() fmt.printf("published message: %sn", message) time.sleep(1 * time.second) } } func generaterandommessage() string { messages := []string{ "hello, world!", "greetings from go!", "mqtt is awesome!", "random message incoming!", "go is fun!", } return messages[rand.intn(len(messages))] } func init() { rand.new(rand.newsource(time.now().unixnano())) }
没什么大不了的,尽管我们可以让它对失败更加稳健。
应用程序创建一个新连接,指定代理 url 和代理将用来识别它的客户端 id。
请注意,发布消息非常简单,只需指定主题、qos、保留标志和消息本身即可。
消息可以是字符串、字节数组或缓冲区。对于qos,我们将其设置为0,这意味着“最多一次”,即不保证传送。
保留标志告诉代理是否将最后一条消息存储到给定主题。如果消息被“保留”,当新客户端订阅给定主题时,最新保留的消息将立即发送给它。在此示例中,我们没有使用该功能。
请注意,此发布者应用程序的源代码可以在此页面底部的链接中找到。
现在让我们看看订阅者端。
package main import ( "context" "fmt" mqtt "github.com/eclipse/paho.mqtt.golang" "os" "os/signal" "sync" "syscall" ) const ( broker = "tcp://localhost:1883" clientID = "go-mqtt-subscriber" topic = "iot-messages" ) var mqttMsgChan = make(chan mqtt.Message) var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) { mqttMsgChan <- msg } func processMsg(ctx context.Context, input <-chan mqtt.Message) chan mqtt.Message { out := make(chan mqtt.Message) go func() { defer close(out) for { select { case msg, ok := <-input: if !ok { return } fmt.Printf("Received message: %s from topic: %sn", msg.Payload(), msg.Topic()) out <- msg case <-ctx.Done(): return } } }() return out } var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) { fmt.Println("Connected to MQTT Broker") } var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) { fmt.Printf("Connection lost: %v", err) } func main() { opts := mqtt.NewClientOptions() opts.AddBroker(broker) opts.SetClientID(clientID) opts.SetDefaultPublishHandler(messagePubHandler) opts.OnConnect = connectHandler opts.OnConnectionLost = connectLostHandler client := mqtt.NewClient(opts) if token := client.Connect(); token.Wait() && token.Error() != nil { panic(token.Error()) } ctx, cancel := context.WithCancel(context.Background()) var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() finalChan := processMsg(ctx, mqttMsgChan) for range finalChan { // just consuming these for now } }() // Subscribe to the topic token := client.Subscribe(topic, 1, nil) token.Wait() fmt.Printf("Subscribed to topic: %sn", topic) // Wait for interrupt signal to gracefully shutdown the subscriber sigChan := make(chan os.Signal, 1) signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM) <-sigChan // Cancel the context to signal the goroutine to stop cancel() // Unsubscribe and disconnect fmt.Println("Unsubscribing and disconnecting...") client.Unsubscribe(topic) client.Disconnect(250) // Wait for the goroutine to finish wg.Wait() fmt.Println("Goroutine terminated, exiting...") }
与发布端类似,应用程序使用代理 url 及其客户端 id 设置客户端。我们设置一个默认处理程序来处理传入的消息;但应用程序订阅的每个主题可能都有不同的主题。
您可以看到此示例包含一些同步来处理终止程序。
还有一个处理消息的管道。虽然对于本例来说不是必需的;当处理大量消息时它会派上用场。
就是这样。从我之前的文章中启动 mqtt 代理或使用另一个代理。您可以在这里找到一个免费的测试经纪人。
在不同的终端中启动发布者和订阅者。您应该会在订阅者控制台中看到收到的消息。
您可以在这里找到发布者代码和订阅者代码
你觉得怎么样?您将如何使出版商变得更加强大?同步和管道可以简化吗?请在下面的评论中告诉我您的想法。