Predix边缘计算的数据总线 - 基于MQTT协议的代理服务器

Predix边缘计算的数据总线 - 基于MQTT协议的代理服务器

Predix Machine容器化架构

Predix Machine容器化架构如下图所示，



从上图中，我们可以了解到，

Predix Machine提供了丰富的工业协议的适配器用来接收设备产生的数据。

Predix Machine还提供了和云端双向通信的能力，既能上传数据到云端，也可以接收云端的控制指令和训练后的分析模型，完成近乎实时的边缘分析。

同时，Predix Machine提供了数据总线和边缘分析应用进行数据交换。

本文我们就详细介绍Predix边缘计算的数据总线 - 基于MQTT协议的代理服务器和MQTT客户端的实践。

MQTT协议

MQTT是专门针对物联网开发的轻量级传输协议，旨在为低带宽和不稳定的网络环境中的物联网设备提供可靠的网络服务。MQTT协议针对低带宽网络，低计算能力的设备，做了特殊的优化，使得其能适应各种物联网应用场景。目前MQTT拥有各种平台和设备上的客户端，已经形成了初步的生态系统。

该协议有如下吸引人的特性：

使用发布/订阅消息模式，提供一对多的消息发布，解除应用程序耦合。

对负载内容屏蔽的消息传输。

使用TCP/IP提供网络连接，相比HTTP协议，MQTT的头部开销非常低，固定头部只有两个字节。

提供三种消息发布服务质量：

至多一次，消息发布完全依赖底层 TCP/IP 网络，会发生消息丢失或重复。这一级别适用于，环境传感器数据，丢失一次读记录无所谓。因为不久后还会有第二次发送。

至少一次，确保消息到达，但消息重复可能会发生。

只有一次，确保消息到达一次。这一级别适用于可靠的消息送达场景，例如计费系统。

MQTT Paho客户端

Predix边缘计算的数据总线就是一个基于MQTT协议的代理服务器，因此，我们只需要通过MQTT协议的客户端就可以订阅数据总线上的相应主题，并获取到数据。MQTT协议的官方网站上推荐使用eclipse paho客户端，因此，我们就用paho的Go语言客户端为例，介绍如何发布和订阅基于MQTT的消息。

下载代码示例

git clone https://github.com/pxie/go.in.practices.git cd go.in.practices/mqtt

搭建本地的MQTT代理服务器

MQTT协议官方网站推荐MQTT代理服务器的一个开源实现项目mosquitto，因此，我们可以用docker非常方便的在本地搭建一个MQTT代理服务器。

docker pull toke/mosquitto
docker run -ti -p 1883:1883 -p 9001:9001 toke/mosquitto

注：docker运行环境的安装，请参考docker官方网站

获取Paho Go语言客户端

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

注：因为官方的MQTT客户端并没有开放获取客户端ID的接口。而为了能在一个进程中更明显的表示不同的客户端，我开发了

GetClientID()

的接口。所以，代码示例中使用的是个人的Paho客户端源码库，而不是官方的客户端源码库。如果大家没有获取客户端ID的需求，强烈建议使用官方的客户端源码库。

连接到MQTT代理服务器

通过paho客户端连接MQTT代理服务器

首先，创建

ClientOptions

对象，并设置相应的连接选项。例如，设置代理服务器的地址，客户端ID等

然后，通过

ClientOptions

对象，创建

Client

对象。

最后，通过

Connect()

方法，使客户端连接到MQTT代理服务器。

// client.go
func (client *client) init(broker string, clientID string) {
opts := mqtt.NewClientOptions()
opts.AddBroker(broker)
opts.SetClientID(clientID)

client.c = mqtt.NewClient(opts)
if token := client.c.Connect(); token.Wait() && token.Error() != nil {
log.Panicln("connect broker error.", token.Error())
}
}

订阅相应的主题并注册回调函数

如果

Client

已经完成了连接，那么订阅相应的主题只需要调用

Subscribe()

方法就可以了。同时paho客户端非常贴心的设计了一个回调函数接口，使我们可以非常方便的在收到相应消息的时候，执行预定的处理函数。在本例中的回调函数只是输出相应的信息到控制台。

//  client.go
func (client *client) sub(topic string, qos byte, callback mqtt.MessageHandler) {
...
if token := client.c.Subscribe(topic, qos, callback); token.Wait() && token.Error() != nil {
log.Panicln("subcribe topic error.", token.Error())
}
}

// 默认的回调函数
func handler(client mqtt.Client, msg mqtt.Message) {
...
log.Printf("default handler. clientID=%s, topic=%s, payload=%v",
client.GetClientID(), msg.Topic(), string(payload))
}

注：订阅方必须先订阅相应的主题，才能收到发布到该主题的消息。如果先发布再订阅，订阅方是不能收到订阅之前所发送的历史消息。仅当发布方将消息的

retain

字段设置为1时，订阅方也只能收到最近一次的历史消息。

发布消息到相应的主题

同订阅主题类似，如果

Client

已经完成了连接，只需要调用

Publish()

方法就可以向相应的主题发布消息了。

// client.go
func (client *client) pub(topic string, qos byte, payload interface{}) {
...
token := client.c.Publish(topic, qos, false, payload)

token.Wait()
if token.Error() != nil {
log.Panicln("publish topic error.", token.Error())
}
}

用goroutine来运行发布和订阅方法

因为我们已经对发布和订阅封装了相应的方法，所以可以用goroutine来并发的处理这些操作。

func main() {
...
go subcriber.sub(topic, qos, handler)

go publisher.pub(topic, qos, createMsg(clientID, 512))

time.Sleep(1 * time.Second)
}

小结

Predix边缘计算的数据总线是基于MQTT协议的代理服务器。

MQTT是专门针对物联网开发的轻量级传输协议，同时提供不同的服务质量。又因为MQTT协议位于TCP之上，所以可以支持远程消息传递。

Paho是MQTT官方网站推荐的客户端，而且有不同语言的实现。

Paho Go客户端已经非常友好的封装了MQTT协议，而且支持goroutine方式的访问。

作者：谢品，上海创新坊首席架构师，GE数字集团

专注于工业互联网，云计算，大数据，高性能分布式存储领域，对Cloud Foundry和传统应用向云端，特别是向Predix迁移有丰富的经验，曾供职于VMware，EMC，Autodesk等知名软件公司云计算部门。