# 下載安裝包
wget https://github.com/emqx/emqx/releases/download/v4.0.4/emqx-ubuntu18.04-v4.0.4.zip
cd mqttd/emqx
.
├── bin
├── data
├── erts-10.5.2
├── etc
├── lib
├── log
└── releases
# 開(kāi)啟服務(wù)
./bin/emqx start
# 查看狀態(tài)
./bin/emqx_ctl status
# 停止服務(wù)
./bin/emqx stop

# 下載依賴包
go get -u github.com/eclipse/paho.mqtt.golang

package main

// 與后端mqtt服務(wù)交互

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

const EMQServerAddress = "你的IP"

// 創(chuàng)建全局mqtt publish消息處理 handler
var messagePubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	fmt.Println("Push Message:")
	fmt.Printf("TOPIC: %s\n", msg.Topic())
	fmt.Printf("MSG: %s\n", msg.Payload())
}

// 創(chuàng)建全局mqtt sub消息處理 handler
var messageSubHandler mqtt.MessageHandler = func(client mqtt.Client, msg mqtt.Message) {
	fmt.Println("收到訂閱消息:")
	fmt.Printf("Sub Client Topic : %s \n", msg.Topic())
	fmt.Printf("Sub Client msg : %s \n", msg.Payload())
}

// 連接的回掉函數(shù)
var connectHandler mqtt.OnConnectHandler =func(client mqtt.Client) {
	fmt.Println("新的連接!" + " Connected")
}

// 丟失連接的回掉函數(shù)
var connectLostHandler mqtt.ConnectionLostHandler = func(client mqtt.Client, err error) {
	fmt.Printf("Connect loss: %v\n", err)
}

func init() {
	// 配置錯(cuò)誤提示
	mqtt.DEBUG = log.New(os.Stdout, "		[mqttDEBUG]", 0)
	mqtt.ERROR = log.New(os.Stdout, " 	[mqttERROR]", 0)
}

/**
 * @Description: 發(fā)布訂閱
 * @param clientID
 * @param addr
 * @param topic
 * @param payload
 */
func Push(topic string, qos byte, retain bool, payload string) {
	// opts ClientOptions 用于設(shè)置 broker，端口，客戶端 id ，用戶名密碼等選項(xiàng)
	opts := mqtt.NewClientOptions().AddBroker("tcp://" + EMQServerAddress + ":1883").SetClientID("test_push")
	opts.SetKeepAlive(60 * time.Second)
	// Message callback handler，在沒(méi)有任何訂閱時(shí)，發(fā)布端調(diào)用此函數(shù)
	opts.SetDefaultPublishHandler(messagePubHandler)
	opts.SetPingTimeout(1 * time.Second)
	opts.OnConnect = connectHandler
	opts.OnConnectionLost = connectLostHandler
	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}
	//發(fā)布消息
	// qos是服務(wù)質(zhì)量: ==1: 一次, >=1: 至少一次, <=1:最多一次
	// retained: 表示mqtt服務(wù)器要保留這次推送的信息，如果有新的訂閱者出現(xiàn)，就會(huì)把這消息推送給它（持久化推送）
	token := client.Publish(topic, qos, retain, payload)
	token.Wait()
	fmt.Println("Push Data : "+topic, "Data Size is "+strconv.Itoa(len(payload)))
	fmt.Println("Disconnect with broker")
	client.Disconnect(250)
}

/**
 * @Description: 訂閱與取消訂閱
 * @param clientID
 * @param addr
 * @param topic
 * @param isSub
 */
func Subscription(topic string, qos byte, isSub bool, handleFun func([]byte)) {
	opts := mqtt.NewClientOptions().AddBroker("tcp://" + EMQServerAddress + ":1883").SetClientID("sub_test")
	opts.SetKeepAlive(60 * time.Second)
	opts.SetPingTimeout(1 * time.Second)
	opts.OnConnect = func(client mqtt.Client) {
		fmt.Println("New Subscription! Connected" + " => " + topic)
	}
	opts.OnConnectionLost = connectLostHandler
	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	if isSub {
		// 訂閱消息
		if token := client.Subscribe(topic, qos, func(client mqtt.Client, msg mqtt.Message) {
			fmt.Printf("Receive Subscribe Message :")
			fmt.Printf("Sub Client Topic : %s, Data size is  %d \n", msg.Topic(), len(msg.Payload()))
			if len(msg.Payload()) > 0 {
				handleFun(msg.Payload())
			}
		}); token.Wait() && token.Error() != nil {
			fmt.Println(token.Error())
			os.Exit(1)
		}
	} else {
		// 取消訂閱
		if token := client.Unsubscribe(topic); token.Wait() && token.Error() != nil {
			fmt.Println(token.Error())
			os.Exit(1)
		}
	}
}