Go安装配置与使用mqtt_go用mqtt开发梯控

1. 安装

1. 引入mosquitto仓库并更新

sudo apt-add-repository ppa:mosquitto-dev/mosquitto-ppa
sudo apt-get update
1
2

2. 执行以下命令安装mosquitto包

sudo apt-get install mosquitto
1

3. 安装mosquitto开发包

sudo apt-get install mosquitto-dev
1

4. 安装mosquitto客户端

sudo apt-get install mosquitto-clients
1

5. 查询mosquitto是否正确运行

sudo service mosquitto status 
1

active则为正常运行

2. 测试

2.1 参数说明

2.2 注册一个top进行接收

mosquitto_sub -h localhost -t "mqtt" -v
1

2.3 发布消息到刚注册的top

需要另开一个终端，当执行完下面命令后，会在上一个终端出打印出我们发布的消息

mosquitto_pub -h localhost -t "mqtt" -m "Hello MQTT"
1

3. 配置Mqtt

设置一下用户名和密码，这样我们的mqtt才会比较安全。mqtt总配置在/etc/mosquitto/mosquitto.conf中

在该文件末尾添加下面三个配置

# 关闭匿名
allow_anonymous false
# 设置用户名和密码
password_file /etc/mosquitto/pwfile
# 配置访问控制列表(topic和用户的关系)
acl_file /etc/mosquitto/acl
# 配置端口
port 8000
1
2
3
4
5
6
7
8

3.1 添加用户

添加一个用户，用户名为pibigstar，这个命令执行之后会让你输入密码，密码自己定义就好

sudo mosquitto_passwd -c /etc/mosquitto/pwfile pibigstar
1

3.2 添加Topic和用户的关系

新增acl文件

sudo vim /etc/mosquitto/acl
1

添加下面内容

# 用户test只能发布以test为前缀的主题
# 订阅以mqtt开头的主题
user test
topic write test/#
topic read mqtt/#

user pibigstar
topic write mqtt/#
topic write mqtt/#
1
2
3
4
5
6
7
8
9

3.3 重启mqtt

sudo /etc/init.d/mosquitto restart
1

3.4 测试

3.4.1 监听消费

mosquitto_sub -h 127.0.0.1 -p 8000 -t "mqtt" -v -u pibigstar -P 123456
1

3.4.2 发布消息

mosquitto_pub -h 127.0.0.1 -p 8000 -t "mqtt" -m "Hello MQTT" -u pibigstar -P 123456
1

4. Go语言使用Mqtt

package mqtt

import (
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	gomqtt "github.com/eclipse/paho.mqtt.golang"
)

const (
	Host     = "192.168.1.101:8000"
	UserName = "pibigstar"
	Password = "123456"
)

type Client struct {
	nativeClient  gomqtt.Client
	clientOptions *gomqtt.ClientOptions
	locker        *sync.Mutex
	// 消息收到之后处理函数
	observer func(c *Client, msg *Message)
}

type Message struct {
	ClientID string `json:"clientId"`
	Type     string `json:"type"`
	Data     string `json:"data,omitempty"`
	Time     int64  `json:"time"`
}

func NewClient(clientId string) *Client {
	clientOptions := gomqtt.NewClientOptions().
		AddBroker(Host).
		SetUsername(UserName).
		SetPassword(Password).
		SetClientID(clientId).
		SetCleanSession(false).
		SetAutoReconnect(true).
		SetKeepAlive(120 * time.Second).
		SetPingTimeout(10 * time.Second).
		SetWriteTimeout(10 * time.Second).
		SetOnConnectHandler(func(client gomqtt.Client) {
			// 连接被建立后的回调函数
			fmt.Println("Mqtt is connected!", "clientId", clientId)
		}).
		SetConnectionLostHandler(func(client gomqtt.Client, err error) {
			// 连接被关闭后的回调函数
			fmt.Println("Mqtt is disconnected!", "clientId", clientId, "reason", err.Error())
		})

	nativeClient := gomqtt.NewClient(clientOptions)

	return &Client{
		nativeClient:  nativeClient,
		clientOptions: clientOptions,
		locker:        &sync.Mutex{},
	}
}

func (client *Client) GetClientID() string {
	return client.clientOptions.ClientID
}

func (client *Client) Connect() error {
	return client.ensureConnected()
}

// 确保连接
func (client *Client) ensureConnected() error {
	if !client.nativeClient.IsConnected() {
		client.locker.Lock()
		defer client.locker.Unlock()
		if !client.nativeClient.IsConnected() {
			if token := client.nativeClient.Connect(); token.Wait() && token.Error() != nil {
				return token.Error()
			}
		}
	}
	return nil
}

// 发布消息
// retained: 是否保留信息
func (client *Client) Publish(topic string, qos byte, retained bool, data []byte) error {
	if err := client.ensureConnected(); err != nil {
		return err
	}

	token := client.nativeClient.Publish(topic, qos, retained, data)
	if err := token.Error(); err != nil {
		return err
	}

	// return false is the timeout occurred
	if !token.WaitTimeout(time.Second * 10) {
		return errors.New("mqtt publish wait timeout")
	}

	return nil
}

// 消费消息
func (client *Client) Subscribe(observer func(c *Client, msg *Message), qos byte, topics ...string) error {
	if len(topics) == 0 {
		return errors.New("the topic is empty")
	}

	if observer == nil {
		return errors.New("the observer func is nil")
	}

	if client.observer != nil {
		return errors.New("an existing observer subscribed on this client, you must unsubscribe it before you subscribe a new observer")
	}
	client.observer = observer

	filters := make(map[string]byte)
	for _, topic := range topics {
		filters[topic] = qos
	}
	client.nativeClient.SubscribeMultiple(filters, client.messageHandler)

	return nil
}

func (client *Client) messageHandler(c gomqtt.Client, msg gomqtt.Message) {
	if client.observer == nil {
		fmt.Println("not subscribe message observer")
		return
	}
	message, err := decodeMessage(msg.Payload())
	if err != nil {
		fmt.Println("failed to decode message")
		return
	}
	client.observer(client, message)
}

func decodeMessage(payload []byte) (*Message, error) {
	message := new(Message)
	decoder := json.NewDecoder(strings.NewReader(string(payload)))
	decoder.UseNumber()
	if err := decoder.Decode(&message); err != nil {
		return nil, err
	}
	return message, nil
}

func (client *Client) Unsubscribe(topics ...string) {
	client.observer = nil
	client.nativeClient.Unsubscribe(topics...)
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157

4.1 测试

package mqtt

import (
	"encoding/json"
	"fmt"
	"sync"
	"testing"
	"time"
)

func TestMqtt(t *testing.T) {
	var (
		clientId = "pibigstar"
		wg       sync.WaitGroup
	)
	client := NewClient(clientId)
	err := client.Connect()
	if err != nil {
		t.Errorf(err.Error())
	}

	wg.Add(1)
	go func() {
		err := client.Subscribe(func(c *Client, msg *Message) {
			fmt.Printf("接收到消息: %+v \n", msg)
			wg.Done()
		}, 1, "mqtt")

		if err != nil {
			panic(err)
		}
	}()

	msg := &Message{
		ClientID: clientId,
		Type:     "text",
		Data:     "Hello Pibistar",
		Time:     time.Now().Unix(),
	}
	data, _ := json.Marshal(msg)

	err = client.Publish("mqtt", 1, false, data)
	if err != nil {
		panic(err)
	}

	wg.Wait()
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48