gomqtt(嵌入式arm设备)接入物联网平台

#1、设备
任意能跑linux的设备。
当前设备:imx6 内存>=128m 容量>=128m
#2、平台
华为云平台
设备接入服务，开发者中心
#3、设备注册
##1、添加产品
选择添加自定义产品

##2、添加profile定义，插件

#4、注册设备
用官方工具计算出clientid等等连接参数
#5、编写代码
##5.1、主函数

// main
package main

import (
	"fmt"
	"sync"

	"time"

	"./user_mqtt"
	//"./user_nanomsg"
	"./utils"
	mqtt "github.com/eclipse/paho.mqtt.golang"
)

type ConfigRead struct {
	sslUrl   string //
	clientId string
	userName string
	passWord string
	pemAdd   string //证书地址

	pubAdd string
	subAdd string
}

//初始化一些变量
var readMsg ConfigRead
//var internalCommunication user_nanomsg.Nanomsg = new(user_nanomsg.NanomsgOptions)
var device1 user_mqtt.DeviceMsg = new(user_mqtt.DeviceMsgOptions)

//var handler *user_nanomsg.NanomsgOptions
func onReceived(client mqtt.Client, message mqtt.Message) {
	fmt.Printf("Receive topic: %s,  payload: %x n", message.Topic(), string(message.Payload()))
	//internalCommunication.PairSocketSend([]byte(message.Payload()))
}

// pub客户端与服务端断连后，触发重连机制
func onPubConnectionLost(client mqtt.Client, err error) {
	fmt.Println("on pub connect lost, try to reconnect")
	user_mqtt.LoopConnect(client)
	client.Subscribe(readMsg.subAdd, 0, onReceived)
}

//读取INI配置文件
func InitConfigRead(o *ConfigRead) bool {
	ini_parser := utils.IniParser{}
	conf_file_name := "/my-demo/mqtt/init.ini"
	if err := ini_parser.Load(conf_file_name); err != nil {
		conf_file_name = "./init.ini"
		if err := ini_parser.Load(conf_file_name); err != nil {
			//fmt.Printf("try load config file[%s] error[%s]n", conf_file_name, err.Error())
			return false
		}
		//return
	}

	o.sslUrl = ini_parser.GetString("connect_sect", "SslUrl")
	o.clientId = ini_parser.GetString("connect_sect", "ClientId")
	o.userName = ini_parser.GetString("connect_sect", "UserName")
	o.passWord = ini_parser.GetString("connect_sect", "PassWord")
	o.pemAdd = ini_parser.GetString("pem_sect", "Pemaddr")

	o.pubAdd = "/huawei/v1/devices/" + readMsg.userName + "/data/binary"
	o.subAdd = "/huawei/v1/devices/" + readMsg.userName + "/command/binary"
	return true
}

//初始化外部通信
func InitMQTTCommunication() bool {
	var ret bool

	if ret = device1.GetPemTlsConfig(readMsg.pemAdd); ret == false {
		fmt.Println("GetPemTlsConfig is error")
		return false
	}

	device1.SettingConnectMsg(readMsg.sslUrl, readMsg.clientId, readMsg.userName, readMsg.passWord)

	device1.SettingPubaddr(readMsg.pubAdd)
	device1.SettingSubaddr(readMsg.subAdd)

	device1.SettingLostCallback(onPubConnectionLost)
	device1.SettingRecCallback(onReceived)

	if ret = device1.StartConnect(); ret == false {
		fmt.Println("StartConnect is error")
		return false
	}

	return true
}

func main() {

	//初始化配置项
	if InitConfigRead(&readMsg) == false {
		fmt.Println("InitConfigRead is error!")
		return
	}

	if InitMQTTCommunication() == false {
		fmt.Println("InitMQTTCommunication is error!!")
		return
	}

	wait := sync.WaitGroup{}
	wait.Add(1)

	pubmsg := []byte{0x00, 0x01}

	go func() {
		for {
			time.Sleep(10 * time.Second)
			device1.PublishMsg(pubmsg)
			//fmt.Println("PublishMsg:%x!!", pubmsg)
		}
	}()

	wait.Wait()
}

##5.2、连接模块代码

package user_mqtt

import (
	"crypto/tls"
	"crypto/x509"
	"io/ioutil"
	"time"

	"fmt"
	//"sync"

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

type DeviceMsg interface {
	GetPemTlsConfig(string) bool                      //获取pem加密
	SettingConnectMsg(string, string, string, string) //设置连接的参数
	SettingPubaddr(string)
	SettingSubaddr(string)
	SettingLostCallback(func(mqtt.Client, error))
	SettingRecCallback(func(mqtt.Client, mqtt.Message))

	StartConnect() bool
	PublishMsg(interface{}) bool                      //上报数据到发布的地址
}

type DeviceMsgOptions struct {
	sslUrl   string //
	clientId string
	userName string
	passWord string

	pubAdd string
	subAdd string

	pemAdd string //证书地址

	Tlsconfig    *tls.Config
	Mqtthandler  mqtt.Client
	LostCallback func(mqtt.Client, error)
	RecCallback  func(mqtt.Client, mqtt.Message)
}

//获取pem加密
func (o *DeviceMsgOptions) GetPemTlsConfig(filename string) bool {
	o.pemAdd = filename
	caCert, err := ioutil.ReadFile(o.pemAdd)
	if err != nil {
		return false
	}
	caCertPool := x509.NewCertPool()
	caCertPool.AppendCertsFromPEM(caCert)

	// Create the TLS Config with the CA pool and enable Client certificate validation
	tlsConfig := &tls.Config{
		ClientCAs:  caCertPool,
		ClientAuth: tls.RequireAndVerifyClientCert,
		// 单向认证，client不校验服务端证书
		InsecureSkipVerify: true,
	}
	o.TlscOnfig= tlsConfig
	return true
}

//设置连接需要的参数
func (o *DeviceMsgOptions) SettingConnectMsg(sslurl string, clientId string, userName string, passWord string) {
	o.sslUrl = sslurl
	o.clientId = clientId
	o.userName = userName
	o.passWord = passWord
}

//上报数据到订阅的地址
func (o *DeviceMsgOptions) PublishMsg(payload interface{}) bool {

	if len(o.pubAdd) == 0 {
		return false
	}

	o.Mqtthandler.Publish(o.pubAdd, 0, false, payload)
	return true
}

//设置发布的地址
func (o *DeviceMsgOptions) SettingPubaddr(pubadd string) {
	o.pubAdd = pubadd
}

//设置订阅的地址的地址
func (o *DeviceMsgOptions) SettingSubaddr(subadd string) {
	o.subAdd = subadd
}

//设置断开触发的程序
func (o *DeviceMsgOptions) SettingLostCallback(callback func(mqtt.Client, error)) {
	o.LostCallback = callback
}

//设置接收触发的程序
func (o *DeviceMsgOptions) SettingRecCallback(callback func(mqtt.Client, mqtt.Message)) {
	o.RecCallback = callback
}

//初始化服务器
func (o *DeviceMsgOptions) StartConnect() bool {
	if len(o.sslUrl) == 0 || len(o.clientId) == 0 || len(o.userName) == 0 || len(o.passWord) == 0 {
		return false
	}
	opts := mqtt.NewClientOptions().AddBroker(o.sslUrl)
	opts.SetClientID(o.clientId)
	opts.SetUsername(o.userName)
	opts.SetPassword(o.passWord)
	opts.SetProtocolVersion(4)

	if len(o.pemAdd) == 0 {
		return false
	}

	opts.SetTLSConfig(o.Tlsconfig)

	opts.OnConnect= onConnect
	opts.AutoRecOnnect= false
	// 回调函数，客户端与服务端断连后立刻被触发
	opts.OnConnectionLost= o.LostCallback
	o.Mqtthandler = mqtt.NewClient(opts)
	LoopConnect(o.Mqtthandler)

	o.Mqtthandler.Subscribe(o.subAdd, 0, o.RecCallback)
	return true
}

//连接成功调用
func onConnect(client mqtt.Client) {
	fmt.Println("on connect")
}

//检查是否连接
func CheckClientToken(token mqtt.Token) (bool, error) {
	if token.Wait() && token.Error() != nil {
		return false, token.Error()
	}
	return true, nil
}

//断开立即调用
func LoopConnect(client mqtt.Client) {
	for {
		token := client.Connect()
		if rs, err := CheckClientToken(token); !rs {
			fmt.Printf("connect error: %sn", err.Error())
		} else {
			break
		}
		time.Sleep(1 * time.Second)
	}
}

##5.3、配置文件读取代码

package utils

import (
    "gopkg.in/ini.v1"
)

type IniParser struct {
    conf_reader *ini.File // config reader
}

type IniParserError struct {
    error_info string
}

func (e *IniParserError) Error() string { return e.error_info }

func (this *IniParser) Load(config_file_name string) error {
    conf, err := ini.Load(config_file_name)
    if err != nil {
        this.conf_reader = nil
        return err
    }
    this.conf_reader = conf
    return nil
}

func (this *IniParser) GetString(section string, key string) string {
    if this.conf_reader == nil {
        return ""
    }

    s := this.conf_reader.Section(section)
    if s == nil {
        return ""
    }

    return s.Key(key).String()
}

func (this *IniParser) GetInt32(section string, key string) int32 {
    if this.conf_reader == nil {
        return 0
    }

    s := this.conf_reader.Section(section)
    if s == nil {
        return 0
    }

    value_int, _ := s.Key(key).Int()

    return int32(value_int)
}

func (this *IniParser) GetUint32(section string, key string) uint32 {
    if this.conf_reader == nil {
        return 0
    }

    s := this.conf_reader.Section(section)
    if s == nil {
        return 0
    }

    value_int, _ := s.Key(key).Uint()

    return uint32(value_int)
}

func (this *IniParser) GetInt64(section string, key string) int64 {
    if this.conf_reader == nil {
        return 0
    }

    s := this.conf_reader.Section(section)
    if s == nil {
        return 0
    }

    value_int, _ := s.Key(key).Int64()
    return value_int
}

func (this *IniParser) GetUint64(section string, key string) uint64 {
    if this.conf_reader == nil {
        return 0
    }

    s := this.conf_reader.Section(section)
    if s == nil {
        return 0
    }

    value_int, _ := s.Key(key).Uint64()
    return value_int
}

func (this *IniParser) GetFloat32(section string, key string) float32 {
    if this.conf_reader == nil {
        return 0
    }

    s := this.conf_reader.Section(section)
    if s == nil {
        return 0
    }

    value_float, _ := s.Key(key).Float64()
    return float32(value_float)
}

func (this *IniParser) GetFloat64(section string, key string) float64 {
    if this.conf_reader == nil {
        return 0
    }

    s := this.conf_reader.Section(section)
    if s == nil {
        return 0
    }

    value_float, _ := s.Key(key).Float64()
    return value_float
}

#6、运行代码


#7、交叉编译
gox -os “linux” -arch arm
传到开发板上运行