首页
学习
活动
专区
工具
TVP
发布
精选内容/技术社群/优惠产品,尽在小程序
立即前往

Robot OS网络通信MQTT实战

1. 背景

最近开发的机器人操作系统 ROS 基于 Android,在里面做一些深度定制,其中运动控制与 Server 的交互需要双向通道,经过权衡和讨论我们最终选用 MQTT 作为长连接通信方案。

2. MQTT 介绍

MQTT 的全称为 Message Queue Telemetry Transport(消息队列遥测传输协议),是 ISO 标准(ISO/IEC PRF 20922)下基于客户端-服务器的消息发布/订阅传输协议,目的是为低带宽和不稳定的网络环境中的物联网设备提供可靠的网络服务。目前在 Iot 应用广泛,主要有以下优点:

  1. 简单:MQTT 是一种消息队列协议,使用发布/订阅消息模式,提供一对多的消息发布,解除应用程序耦合,相对于其他协议,开发更简单;
  2. 传输更加稳定:因为是工作在 TCP/IP 协议上,由 TCP/IP 协议提供稳定的网络连接;
  3. 轻量级:小型传输,开销很小(固定长度的头部是 2 字节),协议交换最小化,降低网络流量;适合低带宽,数据量较小的应用;
  4. 易于实现:MQTT 协议的服务端程序已经非常成熟,PHP,JAVA,Python,C,C#等系统语言都可以实现 MQTT 服务,客户端也同样丰富,eclipse 开源了主流编程语言的 MQTT 客户端实现;
  5. 开放性:因为基于 ISO 标准,所以 MQTT 只提供标准,我们自己可以基于协议实现这个标准,而且可以和别的系统进行对接,加上市面上大量开放源代码,也进一步推动了 MQTT 的发展,百度云、阿里云、中国移动 onenet 等几乎所有的开放性物联网平台都支持 MQTT;

总结下来就是:简单易用开放

3. MQTT 协议分析

3.1 MQTT 设计特点

为了满足低电量消耗和低网络带宽的需求,MQTT 协议在设计之初就包含了以下一些特点:

  1. 实现简单
  2. 提供数据传输的 QoS
  3. 轻量、占用带宽低
  4. 可传输任意类型的数据
  5. 可保持的会话(session)

3.2 MQTT 协议特点

  1. 基于 TCP 协议的应用层协议;
  2. 采用 C/S 架构;
  3. 使用订阅/发布模式,将消息的发送方和接受方解耦;
  4. 提供 3 种消息的 QoS(Quality of Service): 至多一次,最少一次,只有一次;
  5. 收发消息都是异步的,发送方不需要等待接收方应答。

3.3 MQTT 通信模型

MQTT 的通信是通过发布/订阅的方式来实现的,订阅和发布又是基于主题(Topic)的。发布方和订阅方不直接进行连接,而是用到了一个中间方,它们是通过这种方式来进行解耦。两端通信的主要流程如下:

  1. 发布方(Publisher)连接到 Broker;
  2. 订阅方(Subscriber)连接到 Broker,并订阅主题 Topic1;
  3. 发布方(Publisher)发送给 Broker 一条消息,主题为 Topic1;
  4. Broker 收到了发布方的消息,发现订阅方(Subscriber)订阅了 Topic1,然后将消息转发给订阅方(Subscriber);
  5. 订阅方从 Broker 接收该消息;

MQTT 还支持离线消息,发布方在发布消息时并不需要订阅方也连接到 Broker,只要订阅方之前订阅过相应主题,那么它在连接到 Broker 之后就可以收到发布方在它离线期间发布的消息。

3.3 MQTT 协议数据包

MQTT 协议数据包的消息格式为:固定头|可变头|消息体。

  • 固定头(Fixed header):存在于所有的 MQTT 数据包中,用于表示数据包类型及对应标志、数据包大小等;
  • 可变头(Variable header):存在于部分类型的 MQTT 数据包中,具体内容是由相应类型的数据包决定的;
  • 消息体(Payload):存在于部分的 MQTT 数据包中,存储消息的具体数据。

具体的结构说明可以参考官方文档,这里截图两个 MQTT 协议传输时数据内容:

在我们了解 MQTT 开源源码或者自己实现 MQTT 协议的库时再着重研究协议具体内容。

4. MQTT 开发环境搭建

4.1 搭建 MQTT 服务

我们可以使用 mosquitto 快速搭建 mqtt server。在 ubuntu 上直接执行sudo apt-get install mosquitto,在 Mac 上执行sudo brew install mosquitto即可。

具体配置操作以 ubuntu 为例:

sudo service mosquitto statussudo service mosquitto startsudo service mosquitto stop

conf 配置文件 mosquitto.conf:

pid_file /var/run/mosquitto.pid

# 消息持久存储
persistence true
persistence_location /var/lib/mosquitto/

# 日志文件
log_dest file /var/log/mosquitto/mosquitto.log

# 其他配置
include_dir /etc/mosquitto/conf.d

# 禁止匿名访问
allow_anonymous false
# 认证配置
password_file /etc/mosquitto/pwfile
# 权限配置
acl_file /etc/mosquitto/aclfile

mosquitto -c /etc/mosquitto/mosquitto.conf -d

  • -c 指定配置文件
  • -d 表示后台启动

4.2 验证服务

安装 mosquitto 配套客户端文件:

sudo apt-get install mosquitto-clients

mosquitto-client 包含 pub 和 sub 两个命令工具,对应参数说明:

mosquitto_pub 命令参数说明:

  • -d 打印 debug 信息
  • -f 将指定文件的内容作为发送消息的内容
  • -h 指定要连接的域名 默认为 localhost
  • -i 指定要给哪个 clientId 的用户发送消息
  • -I 指定给哪个 clientId 前缀的用户发送消息
  • -m 消息内容
  • -n 发送一个空(null)消息
  • -p 连接端口号
  • -q 指定 QoS 的值(0,1,2)
  • -t 指定 topic
  • -u 指定 broker 访问用户
  • -P 指定 broker 访问密码
  • -V 指定 MQTT 协议版本
  • --will-payload 指定一个消息,该消息当客户端与 broker 意外断开连接时发出。该参数需要与--will-topic 一起使用
  • --will-qos Will 的 QoS 值。该参数需要与--will-topic 一起使用
  • --will-retain 指定 Will 消息被当做一个 retain 消息(即消息被广播后,该消息被保留起来)。该参数需要与--will-topic 一起使用
  • --will-topic 用户发送 Will 消息的 topic

mosquitto_sub 命令参数说明:

  • -c 设定‘clean session’为无效状态,这样一直保持订阅状态,即便是已经失去连接,如果再次连接仍旧能够接收的断开期间发送的消息。
  • -d 打印 debug 信息
  • -h 指定要连接的域名 默认为 localhost
  • -i 指定 clientId
  • -I 指定 clientId 前缀
  • -k keepalive 每隔一段时间,发 PING 消息通知 broker,仍处于连接状态。 默认为 60 秒。
  • -q 指定希望接收到 QoS 为什么的消息 默认 QoS 为 0
  • -R 不显示陈旧的消息
  • -t 订阅 topic
  • -v 打印消息
  • --will-payload 指定一个消息,该消息当客户端与 broker 意外断开连接时发出。该参数需要与--will-topic 一起使用
  • --will-qos Will 的 QoS 值。该参数需要与--will-topic 一起使用
  • --will-retain 指定 Will 消息被当做一个 retain 消息(即消息被广播后,该消息被保留起来)。该参数需要与--will-topic 一起使用
  • --will-topic 用户发送 Will 消息的 topic

示例:

mosquitto_pub -h localhost -p 1883 -t "demo/1" -m "test"

mosquitto_sub -h localhost -p 1883 -t "demo/1"

直接运行上面命令既可完成发布与订阅操作。

4.3 跑通 Android 平台客户端

上面是 mosquitto 提供的现成的 PC 命令工具,我们要在 Android 平台上跑通客户端需要用 Java 和 C++实现一套,eclipse 也提供了 C++/C 以及 Java 版本的库,我们直接编译到 Android 平台即可。已 C 库为例:

下载https://github.com/eclipse/paho.mqtt.c源码,创建 AndroidStudio 工程后将下载的源码拷贝到 cpp 目录下,编写 CMakelist.txt 脚本:

make_minimum_required(VERSION 3.4.1)

set(CMAKE_INSTALL_PREFIX "${CMAKE_BINARY_DIR}" CACHE PATH "Installation directory" FORCE)
message(STATUS "CMAKE_INSTALL_PREFIX=${CMAKE_INSTALL_PREFIX}")

project(mqtt)
SET(VERSION 0.0.1)
add_definitions(-w)
#file(READ version.major PAHO_VERSION_MAJOR)
#file(READ version.minor PAHO_VERSION_MINOR)
#file(READ version.patch PAHO_VERSION_PATCH)
SET(CLIENT_VERSION ${PAHO_VERSION_MAJOR}.${PAHO_VERSION_MINOR}.${PAHO_VERSION_PATCH})

STRING(TIMESTAMP BUILD_TIMESTAMP UTC)
MESSAGE(STATUS "Timestamp is ${BUILD_TIMESTAMP}")

## build options
SET(PAHO_WITH_SSL FALSE CACHE BOOL "Flag that defines whether to build ssl-enabled binaries too. ")
SET(PAHO_BUILD_SHARED TRUE CACHE BOOL "Build shared library")
SET(PAHO_BUILD_STATIC FALSE CACHE BOOL "Build static library")
SET(PAHO_BUILD_DOCUMENTATION FALSE CACHE BOOL "Create and install the HTML based API documentation (requires Doxygen)")
SET(PAHO_BUILD_SAMPLES FALSE CACHE BOOL "Build sample programs")
SET(PAHO_BUILD_DEB_PACKAGE FALSE CACHE BOOL "Build debian package")
SET(PAHO_ENABLE_TESTING TRUE CACHE BOOL "Build tests and run")
SET(PAHO_ENABLE_CPACK TRUE CACHE BOOL "Enable CPack")
SET(PAHO_HIGH_PERFORMANCE FALSE CACHE BOOL "Disable tracing and heap tracking")
SET(PAHO_USE_SELECT FALSE CACHE BOOL "Revert to select system call instead of poll")

IF (PAHO_HIGH_PERFORMANCE)
    ADD_DEFINITIONS(-DHIGH_PERFORMANCE=1)
ENDIF()

IF (PAHO_USE_SELECT)
    ADD_DEFINITIONS(-DUSE_SELECT=1)
ENDIF()

IF (NOT PAHO_BUILD_SHARED AND NOT PAHO_BUILD_STATIC)
    MESSAGE(FATAL_ERROR "You must set either PAHO_BUILD_SHARED, PAHO_BUILD_STATIC, or both")
ENDIF()


SET(ROOT ${CMAKE_SOURCE_DIR})
SET(BUILD_PATH "${CMAKE_BINARY_DIR}")
MESSAGE(STATUS "current project name is ${PROJECTNAME}")

SET(CMAKE_BUILD_TYPE "DEBUG")
#SET(CMAKE_BUILD_TYPE "RELEASE")

set(SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/src/main/cpp/paho.mqtt.c/src)
set(JNI_SOURCE_DIR ${CMAKE_CURRENT_SOURCE_DIR}/src/main/cpp)


find_library(log-lib log)

include_directories(${SOURCE_DIR})
include_directories(${JNI_SOURCE_DIR})

set(common_src
        ${SOURCE_DIR}/MQTTTime.c
        ${SOURCE_DIR}/MQTTProtocolClient.c
        ${SOURCE_DIR}/Clients.c
        ${SOURCE_DIR}/utf-8.c
        ${SOURCE_DIR}/MQTTPacket.c
        ${SOURCE_DIR}/MQTTPacketOut.c
        ${SOURCE_DIR}/Messages.c
        ${SOURCE_DIR}/Tree.c
        ${SOURCE_DIR}/Socket.c
        ${SOURCE_DIR}/Log.c
        ${SOURCE_DIR}/MQTTPersistence.c
        ${SOURCE_DIR}/Thread.c
        ${SOURCE_DIR}/MQTTProtocolOut.c
        ${SOURCE_DIR}/MQTTPersistenceDefault.c
        ${SOURCE_DIR}/SocketBuffer.c
        ${SOURCE_DIR}/LinkedList.c
        ${SOURCE_DIR}/MQTTProperties.c
        ${SOURCE_DIR}/MQTTReasonCodes.c
        ${SOURCE_DIR}/Base64.c
        ${SOURCE_DIR}/SHA1.c
        ${SOURCE_DIR}/WebSocket.c
        ${SOURCE_DIR}/Proxy.c
        ${SOURCE_DIR}/MQTTClient.c
        ${JNI_SOURCE_DIR}/JNI_OnLoad.cc
        ${JNI_SOURCE_DIR}/jni_utils.cc
        ${JNI_SOURCE_DIR}/keutil.cc
        )
IF (NOT PAHO_HIGH_PERFORMANCE)
    SET(common_src ${common_src}
            ${SOURCE_DIR}/StackTrace.c
            ${SOURCE_DIR}/Heap.c
            )
ENDIF()

SET(LIBS_SYSTEM c dl)


set(SELF_LIB_NAME kemqtt)



add_library(${SELF_LIB_NAME} SHARED ${common_src})

target_link_libraries(${SELF_LIB_NAME}
        ${log-lib}
        )

封装 JNI 接口进行 Pub:

static jlong
_start(JNIEnv *env, jclass cls) {
    LOGI("start...");
  MQTTClient client;
  MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
  conn_opts.username = "A1_TEST_TOKEN";
  MQTTClient_message pubmsg = MQTTClient_message_initializer;
  MQTTClient_deliveryToken token;
  int rc;

  if ((rc = MQTTClient_create(&client, ADDRESS, CLIENTID,
                MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS)
  {
  LOGI("Failed to create client, return code %d\n", rc);
  return -1;
  }

  conn_opts.keepAliveInterval = 20;
  conn_opts.cleansession = 1;
  if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
  {
  LOGI("Failed to connect, return code %d\n", rc);
  return -2;
  }

  pubmsg.payload = (void *)PAYLOAD;
  pubmsg.payloadlen = (int)strlen(PAYLOAD);
  pubmsg.qos = QOS;
  pubmsg.retained = 0;
  if ((rc = MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token)) != MQTTCLIENT_SUCCESS)
  {
  LOGI("Failed to publish message, return code %d\n", rc);
  return -3;
  }

  LOGI("Waiting for up to %d seconds for publication of %s\n"
     "on topic %s for client with ClientID: %s\n",
     (int)(TIMEOUT/1000), PAYLOAD, TOPIC, CLIENTID);
  rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
  LOGI("Message with delivery token %d delivered\n", token);

  if ((rc = MQTTClient_disconnect(client, 10000)) != MQTTCLIENT_SUCCESS)
  LOGI("Failed to disconnect, return code %d\n", rc);
  MQTTClient_destroy(&client);
    return (jlong) 0;
}

这里直接改写官方 demo,最开始怎么也连不上,连接错误返回-1,最后发现是测试机好久没用,没有联网了,浪费了好长时间去查代码问题;网络修复后又遇到返回错误 5 的问题,是因为 server 开启了用户验证,所以需要配置 username,配置好后成功完成了发布。

5. 参考资源

  1. MQTT Client各语言实现大全
  2. MQTT协议报文格式解析

6. 总结

本文介绍了 MQTT 和 MQTT 协议的数据包结构,并且介绍了 MQTT 开发环境的搭建,都比较粗浅,后续文章深入分析 MQTT 协议内容以及 MQTT Java 和 C 版本代码实现细节,从代码角度分析 MQTT 协议的优点,并完成实现 MQTT 的 Android 版本交互以及与 Protbuf 的结合等。

  • 发表于:
  • 本文为 InfoQ 中文站特供稿件
  • 首发地址https://www.infoq.cn/article/1370c05c7095d4b73b1e79ef3
  • 如有侵权,请联系 cloudcommunity@tencent.com 删除。

扫码

添加站长 进交流群

领取专属 10元无门槛券

私享最新 技术干货

扫码加入开发者社群
领券