MQTT　golang 客户端

支持

• 支持mqtt 3.1.1协议
• 支持同步模式
• 支持事件模式
• 支持自动重连
• 支持报文数量统计，支持tcp tls连接
• 目前仅支持内存数据存储，可以扩展其他存储

获取

$ go get github.com/antlinker/go-mqtt/client

例子

• 参考例子项目mqttgroup的go-mqtt分支
• 参考例子项目mqttpersonal的go-mqtt分支

API使用

客户端创建

	import (
		fmt
		 "github.com/antlinker/go-mqtt/client"
	)
	.
	.
	.
	//开始日志　false则关闭日志显示
	client.Mlog.SetEnabled(true)
	client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:8883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	.
	.
	.
	//断开连接
	client.Disconnect()

连接事件

	//连接事件接口
	type MqttDisConnListener interface {
		event.Listener
		OnDisconning(event *MqttEvent)
		OnDisconned(event *MqttEvent)
		OnLostConn(event *MqttEvent, err error)
	}
	//默认连接事件实现，可以打印事件信息
	type DefaultPrintDisConnListen struct {
	}
	
	func (l *DefaultPrintDisConnListen) OnLostConn(event *MqttEvent, err error) {
		Mlog.Debugf("OnLostConn :%v", err)
	}
    
	func (l *DefaultPrintDisConnListen) OnDisconning(event *MqttEvent) {
		Mlog.Debugf("OnDisconning")
	}
	func (l *DefaultPrintDisConnListen) OnDisconned(event *MqttEvent) {
	 	Mlog.Debugf("OnDisconned")
	}
	//默认事件实现，不做任何事情
	type DefaultDisConnListen struct {
	}

	func (l *DefaultDisConnListen) OnDisconning(event *MqttEvent) {

	}
	func (l *DefaultDisConnListen) OnDisconned(event *MqttEvent) {

	}
	func (l *DefaultDisConnListen) OnLostConn(event *MqttEvent, err error) {

	}
    

例子

    .
    .
    .
	 client, err := client.CreateClient(client.MqttOption{
		Addr:  "tcp://localhost:1883",
		//断开连接１秒后自动连接，０不自动重连
		ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintDisConnListen{}
	//注册事件
	client.AddConnListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	//移除事件
	client.RemoveConnListener(listener)   
	.
	.
	.

断开连接事件

	//事件接口
	type MqttDisConnListener interface {
		event.Listener
		OnDisconning(event *MqttEvent)
		OnDisconned(event *MqttEvent)
		OnLostConn(event *MqttEvent, err error)
	}
	//默认事件实现，可以打印事件信息
	type DefaultPrintDisConnListen struct {
	}
	func (l *DefaultPrintDisConnListen) OnLostConn(event *MqttEvent, err error) {
		Mlog.Debugf("OnLostConn :%v", err)
	}
	func (l *DefaultPrintDisConnListen) OnDisconning(event *MqttEvent) {
		Mlog.Debugf("OnDisconning")
	}
	func (l *DefaultPrintDisConnListen) OnDisconned(event *MqttEvent) {
		Mlog.Debugf("OnDisconned")
	}
	//默认事件实现，不做任何事情
	type DefaultDisConnListen struct {
	}
	func (l *DefaultDisConnListen) OnDisconning(event *MqttEvent) {
	}
	func (l *DefaultDisConnListen) OnDisconned(event *MqttEvent) {
	}
	func (l *DefaultDisConnListen) OnLostConn(event *MqttEvent, err error) {
	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
		Addr:  "tcp://localhost:1883",
		//断开连接１秒后自动连接，０不自动重连
		ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintDisConnListen{}
	//注册断开连接事件
	client.AddDisConnListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	client.Disconnect()
	//移除事件
	client.RemoveDisConnListener(listener)   
	.
	.
	.

订阅事件

	//事件接口
	type MqttSubListener interface {
		event.Listener
		OnSubStart(event *MqttEvent, sub []SubFilter)
		OnSubSuccess(event *MqttEvent, sub []SubFilter, result []QoS)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintSubscribeListen struct {
	}
	func (*DefaultPrintSubscribeListen) OnSubStart(event *MqttEvent, sub []SubFilter) {
		Mlog.Debugf("OnSubStart:%v", sub)
	}
	func (*DefaultPrintSubscribeListen) OnSubSuccess(event *MqttEvent, sub []SubFilter, result []QoS) {
		Mlog.Debugf("OnSubSuccess:%v:%v", sub, result)
	}
	//默认事件实现，不做任何事情
	type DefaultSubscribeListen struct {
	}
	func (*DefaultSubscribeListen) OnSubStart(event *MqttEvent, sub []SubFilter) {
	}
	func (*DefaultSubscribeListen) OnSubSuccess(event *MqttEvent, sub []SubFilter, result []QoS) {
	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintSubscribeListen{}
	//注册订阅事件
	client.AddSubListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq,err:= client.Subscribe("Test/1", client.QoS1)
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",err))
	}
	
	//等待订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
	}
	mq,err= client.Subscribes(client.CreateSubFilter("test/1", QoS1),
		client.CreateSubFilter("test/2", QoS2),
		client.CreateSubFilter("test/0", QoS0))
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",err))
	}
	//等待订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
	}
	.
	.
	.
	//移除事件
	client.RemoveSubListener(listener)   
	.
	.
	.

发布消息事件

	//事件接口
	//发布消息事件监听
	type MqttPublishListener interface {
		event.Listener
		//准备发送消息
		OnPubReady(event *MqttPublishEvent, packet *MqttPacket)
		//消息发布成功
		OnPubSuccess(event *MqttPublishEvent, packet *MqttPacket)
		//消息发布完成
		OnPubFinal(event *MqttPublishEvent, packet *MqttPacket)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintPublishListen struct {
	}
	func (*DefaultPrintPublishListen) OnPubReady(event *MqttPublishEvent, mp *MqttPacket) {
		Mlog.Debugf("OnPubReady:%v", event.GetSendCnt(PubCnt_TOTAL))
	}
	func (*DefaultPrintPublishListen) OnPubSuccess(event *MqttPublishEvent, mp *MqttPacket) {
		Mlog.Debugf("OnPubSuccess:%v", mp.Packet)

	}
	func (*DefaultPrintPublishListen) OnPubFinal(event *MqttPublishEvent, mp *MqttPacket) {
		Mlog.Debugf("OnPubFinal:%v", mp.Packet)
	}

	//默认事件实现，不做任何事情
	type DefaultPublishListen struct {
	}

	func (*DefaultPublishListen) OnPubReady(event *MqttPublishEvent, mp *MqttPacket) {

	}
	func (*DefaultPublishListen) OnPubSuccess(event *MqttPublishEvent, mp *MqttPacket) {

	}
	func (*DefaultPublishListen) OnPubFinal(event *MqttPublishEvent, mp *MqttPacket) {

	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintPublishListen{}
	//注册订阅事件
	client.AddPubListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq, err := client.Publish("test0", QoS0, false, []byte("测试"))
	if err!=nil {
		//发布消息失败
		panic(fmt.Errorf("发布消息失败:%v",err))
	}
	
	//等待发布消息完成
	mq.Wait()
	if mq.Err()!=nil {
		//发布消息失败
		panic(fmt.Errorf("发布消息失败:%v",mqt.Err()))
	}
	
	.
	.
	.
	//移除事件
	client.RemoveSubListener(listener)   
	.
	.
	.

接收消息事件

	//事件接口
	//接收消息事件监听
	type MqttRecvPubListener interface {
		event.Listener
		OnRecvPublish(event *MqttRecvPubEvent, topic string, payload []byte, qos QoS)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintRecvPubListen struct {
	}
	func (l *DefaultPrintRecvPubListen) OnRecvPublish(event *MqttRecvPubEvent, topic string, payload []byte, qos QoS) {
		Mlog.Debugf("OnRecvPublish:%s(%d) :%s", topic, qos, string(payload))
	}
	//默认事件实现，不做任何事情
	type DefaultRecvPubListen struct {
	}

	func (l *DefaultRecvPubListen) OnRecvPublish(event *MqttRecvPubEvent, topic string, payload []byte, qos QoS) {

	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&MqttRecvPubListener{}
	//注册订阅事件
	client.AddRecvPubListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq,err= client.Subscribe("test",QoS2)
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",err))
	}
	//等待订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
	}
	mq, err := client.Publish("test", QoS1, false, []byte("测试"))
	if err!=nil {
		//发布消息失败
		panic(fmt.Errorf("发布消息失败:%v",err))
	}
	
	//等待发布消息完成
	mq.Wait()
	if mq.Err()!=nil {
		//发布消息失败
		panic(fmt.Errorf("发布消息失败:%v",mqt.Err()))
	}
	
	.
	.
	.
	//移除事件
	client.RemoveRecvPubListener(listener)   
	.
	.
	.

取消订阅事件

	//事件接口
	type MqttUnSubListener interface {
		event.Listener
		OnUnSubStart(event *MqttEvent, filter []string)
		OnUnSubSuccess(event *MqttEvent, filter []string)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintUnSubListen struct {
	}

	func (*DefaultPrintUnSubListen) OnUnSubStart(event *MqttEvent, filter []string) {
		Mlog.Debugf("OnUnSubStart:%v", filter)
	}
	func (*DefaultPrintUnSubListen) OnUnSubSuccess(event *MqttEvent, filter []string) {
		Mlog.Debugf("OnUnSubSuccess:%v", filter)
	}

	//默认事件实现，不做任何事情
	type DefaultUnSubListen struct {
	}
	func (*DefaultUnSubListen) OnUnSubStart(event *MqttEvent, filter []string) {
	}
	func (*DefaultUnSubListen) OnUnSubSuccess(event *MqttEvent, filter []string) {
	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintUnSubListen{}
	//注册订阅事件
	client.AddUnSubListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq,err:= client.UnSubscribe("Test/1")
	if err!=nil {
		//取消订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	
	//等待取消订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//取消订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",mqt.Err()))
	}
	mq,err= client.UnSubscribes("test1","test2")
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	//等待取消订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//订阅失败
		panic(fmt.Errorf("订阅失败:%v",mqt.Err()))
	}
	.
	.
	.
	//移除事件
	client.RemoveUnSubListener(listener)   
	.
	.
	.

报文发送接收事件

	//发送接收报文接口
	type MqttPacketListener interface {
		event.Listener
		//接收到报文
		OnRecvPacket(event *MqttEvent, msg packet.MessagePacket, recvPacketCnt int64)
		//发送报文
		OnSendPacket(event *MqttEvent, msg packet.MessagePacket, sendPacketCnt int64, err error)
	}
    	//默认事件实现，可以打印事件信息
	type DefaultPrintPacketListen struct {
	}
	func (l *DefaultPrintPacketListen) OnRecvPacket(event *MqttEvent, msg packet.MessagePacket, recvPacketCnt int64) {
		Mlog.Debugf("OnRecvPacket:(%d) :%v", recvPacketCnt, msg)
	}
	func (l *DefaultPrintPacketListen) OnSendPacket(event *MqttEvent, msg packet.MessagePacket, sendPacketCnt int64, err error) {
		Mlog.Debugf("OnSendPacket:(%d) :%v\n %v", sendPacketCnt, msg, err)
	}

	//默认事件实现，不做任何事情
	type DefaultPacketListen struct {
	}
	func (l *DefaultPacketListen) OnRecvPacket(event *MqttEvent, packet packet.MessagePacket, recvPacketCnt int64) {
	}
	func (l *DefaultPacketListen) OnSendPacket(event *MqttEvent, packet packet.MessagePacket, sendPacketCnt int64) {
	}
    

例子

	.
	.
	.
	 client, err := client.CreateClient(client.MqttOption{
			Addr:  "tcp://localhost:1883",
			//断开连接１秒后自动连接，０不自动重连
			ReconnTimeInterval: 1,
	})
	if err!=nil {
		//配置文件解析失败
		panic("配置文件解析失败")
	}
	listener:=&DefaultPrintPacketListen{}
	//注册订阅事件
	client.AddPacketListener(listener)
	//建立连接
	err:=client.Connect()
	if err!=nil {
		//连接失败，不会进入自动重连状态
		panic(fmt.Errorf("连接失败:%v",err))
	}
	mq,err:= client.Subscribe("Test/1",client.QoS1)
	if err!=nil {
		//取消订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	
	//等待取消订阅成功
	mq.Wait()
	if mq.Err()!=nil {
		//取消订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",mqt.Err()))
	}
	mq,err= client.Publish("Test/1", QoS1, false, []byte("测试"))
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	mq,err= client.Publish("Test/2", QoS2, false, []byte("测试"))
	if err!=nil {
		//订阅失败
		panic(fmt.Errorf("取消订阅失败:%v",err))
	}
	.
	.
	.
	//移除事件
	client.RemovePacketListener(listener)   
	.
	.
	.