Golang에서 MQTT를 사용하는 방법

Golang은 Google에서 개발한 정적으로 강력하게 유형이 지정되고 컴파일되고 동시 실행되며 가비지 수집 프로그래밍 언어입니다. Go는 표현력이 풍부하고 깔끔하며 효율적입니다. 동시성 메커니즘을 통해 멀티코어 및 네트워크 머신의 사용을 극대화하는 프로그램을 쉽게 작성할 수 있으며 혁신적인 유형 시스템은 유연하고 모듈화된 프로그램 구성을 가능하게 합니다. Go는 기계 코드로 빠르게 컴파일되지만 가비지 수집의 편리함과 런타임 리플렉션의 힘이 있습니다. 동적으로 유형이 지정되고 해석되는 언어와 같이 빠르고 정적으로 유형이 지정되고 컴파일된 언어입니다.

MQTT은 게시/구독 모델을 기반으로 하는 일종의 경량 IoT 메시징 프로토콜로, 매우 적은 코드와 대역폭만 사용하여 IoT 장치에 대한 실시간 및 안정적인 메시징 서비스를 제공할 수 있습니다. 제한된 하드웨어 리소스를 가진 장치와 제한된 대역폭을 가진 네트워크 환경에 적합합니다. 따라서 MQTT protocol은 IoT, 모바일 인터넷, IoV, 전력 및 기타 산업에서 널리 사용됩니다.

이 기사는 주로 Golang 프로젝트에서 paho.mqtt.golang 클라이언트 라이브러리를 사용하는 방법과 클라이언트와 MQTT broker 간의 연결, 구독 및 메시징을 구현하는 방법을 소개합니다.

프로젝트 초기화

이 프로젝트는 go 1.13.12를 기반으로 개발하고 테스트합니다.

go version
go version go1.13.12 darwin/amd64

이 프로젝트는 paho.mqtt.golang을 MQTT 클라이언트 라이브러리로 사용하고 다음을 설치합니다.

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

Go MQTT 사용

이 글은 EMQX에서 제공하는 free public MQTT broker을 사용합니다. 이 서비스는 EMQX의 MQTT IoT cloud platform을 기반으로 만들어집니다. 서버의 접속 정보는 다음과 같습니다.

브로커: broker.emqx.io

TCP 포트: 1883

웹 소켓 포트: 8083

MQTT 브로커에 연결

package main

import (
    "fmt"
    mqtt "github.com/eclipse/paho.mqtt.golang"
    "time"
)

var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
}

var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
    fmt.Println("Connected")
}

var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
    fmt.Printf("Connect lost: %v", err)
}

func main() {
    var broker = "broker.emqx.io"
    var port = 1883
    opts := mqtt.NewClientOptions()
    opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))
    opts.SetClientID("go_mqtt_client")
    opts.SetUsername("emqx")
    opts.SetPassword("public")
    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())
  }
}

클라이언트 옵션: 브로커, 포트, 클라이언트 ID, 사용자 이름, 암호 및 기타 옵션을 설정하는 데 사용됩니다.

messagePubHandler: 전역 MQTT 게시 메시지 처리

connectHandler: 연결에 대한 콜백

connectLostHandler: 연결 손실에 대한 콜백

TSL 연결을 사용하려면 다음 설정을 사용할 수 있습니다.

func NewTlsConfig() *tls.Config {
    certpool := x509.NewCertPool()
    ca, err := ioutil.ReadFile("ca.pem")
    if err != nil {
        log.Fatalln(err.Error())
    }
    certpool.AppendCertsFromPEM(ca)
    // Import client certificate/key pair
    clientKeyPair, err := tls.LoadX509KeyPair("client-crt.pem", "client-key.pem")
    if err != nil {
        panic(err)
    }
    return &tls.Config{
        RootCAs: certpool,
        ClientAuth: tls.NoClientCert,
        ClientCAs: nil,
        InsecureSkipVerify: true,
        Certificates: []tls.Certificate{clientKeyPair},
    }
}

클라이언트 인증서가 설정되지 않은 경우 다음과 같이 설정할 수 있습니다.

func NewTlsConfig() *tls.Config {
    certpool := x509.NewCertPool()
    ca, err := ioutil.ReadFile("ca.pem")
    if err != nil {
        log.Fatalln(err.Error())
    }
    certpool.AppendCertsFromPEM(ca)
    return &tls.Config{
        RootCAs: certpool,
}

그런 다음 TLS를 설정합니다.

var broker = "broker.emqx.io"
var port = 8883
opts := mqtt.NewClientOptions()
opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))
tlsConfig := NewTlsConfig()
opts.SetTLSConfig(tlsConfig)
// other options

신청

func sub(client mqtt.Client) {
    topic := "topic/test"
    token := client.Subscribe(topic, 1, nil)
    token.Wait()
    fmt.Printf("Subscribed to topic %s", topic)
}

메시지 게시

func publish(client mqtt.Client) {
    num := 10
    for i := 0; i < num; i++ {
        text := fmt.Sprintf("Message %d", i)
        token := client.Publish("topic/test", 0, false, text)
        token.Wait()
        time.Sleep(time.Second)
    }
}

테스트

테스트를 위해 다음 코드를 사용합니다.

package main

import (
    "fmt"
    mqtt "github.com/eclipse/paho.mqtt.golang"
    "log"
    "time"
)

var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
    fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
}

var connectHandler mqtt.OnConnectHandler = func(client mqtt.Client) {
    fmt.Println("Connected")
}

var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
    fmt.Printf("Connect lost: %v", err)
}

func main() {
    var broker = "broker.emqx.io"
    var port = 1883
    opts := mqtt.NewClientOptions()
    opts.AddBroker(fmt.Sprintf("tcp://%s:%d", broker, port))
    opts.SetClientID("go_mqtt_client")
    opts.SetUsername("emqx")
    opts.SetPassword("public")
    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())
    }

    sub(client)
    publish(client)

    client.Disconnect(250)
}

func publish(client mqtt.Client) {
    num := 10
    for i := 0; i < num; i++ {
        text := fmt.Sprintf("Message %d", i)
        token := client.Publish("topic/test", 0, false, text)
        token.Wait()
        time.Sleep(time.Second)
    }
}

func sub(client mqtt.Client) {
    topic := "topic/test"
    token := client.Subscribe(topic, 1, nil)
    token.Wait()
  fmt.Printf("Subscribed to topic: %s", topic)
}

코드를 실행하면 MQTT 연결 및 구독이 성공적으로 이루어지고 구독 주제 메시지를 성공적으로 수신할 수 있음을 알 수 있습니다.

요약

지금까지 paho.mqtt.golang 클라이언트를 사용하여 public MQTT broker에 연결하고 테스트 클라이언트와 MQTT 브로커 간의 연결, 메시지 게시 및 구독을 구현했습니다.

IoT 개발 및 MQTT에 대한 더 많은 기사를 게시할 예정이니 계속 지켜봐 주시기 바랍니다.