使用 Go 的简单 MQTT 应用程序

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: %s\n", 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()))
}

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: %s\n", 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: %s\n", 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...")
}