物联网mqtt协议浅谈 (物联网mqtt协议内容)

完整项目：https://github.com/IoT-Technology/IOT-Technical-Guide

物联网mqtt从搭建到实战教程,mqtt物联网使用协议

指南logo

系列文章：

高质量的 IOT 技术教程，代码主要源于国外开源物联网平台ThingsBoard和对阿里云物联网平台的感悟

物联网mqtt从搭建到实战教程,mqtt物联网使用协议

源码解析系列

a.『准备篇』

《物联网时代-Thingsboard源码分析-调试环境调试》
《物联网时代-Thingsboard源码分析-项目结构说明》

b.『设备连接协议篇』

MQTT

协议 MQTT： http://mqtt.org/

技术框架 Netty：https://netty.io/

《MQTT入门篇》
《物联网时代-ThingsBoard源码分析-MQTT设备连接协议-上》

物联网mqtt从搭建到实战教程,mqtt物联网使用协议

MQTT概述

MQTT是一种轻量级的发布-订阅消息传递协议，它可能最适合各种物联网设备。你可以在此处找到有关MQTT的更多信息。

ThingsBoard服务器节点充当支持QoS级别0（最多一次）和QoS级别1（至少一次）以及一组预定义主题的MQTT主题。

ThingsBoard基于MQTT协议提供给设备的API是非常"灵活"的。

例如，目前提供了四种API:

遥测数据上传API
属性API
RPC API
声明设备所有权API

用上述四种的API，用户可以动态调整设备监控属性，例如原先采集设备的温度，湿度及材料大小调整为温度，湿度和材料是否合格；并可以获取设备历史遥测数据和最新遥测数据；通过RPC API, 设备和服务器可以实时获取对方相应属性的变化和通过API用户可以对设备进行转让，租赁和回收操作。

关联模块一览

和MQTT设备传输协议关联的模块有Thingsboard MQTT Transport Service、Thingsboard MQTT Transport Common和Thingsboard Server Queue components。前面这些名称大家可以看IDEA maven模块名称。

物联网mqtt从搭建到实战教程,mqtt物联网使用协议

MQTT Transport Service

org.thingsboard.server.mqtt.ThingsboardMqttTransportApplication，MQTT服务启动类，使用SpringBoot启动类，通过加载模块和配置文件，对服务进行配置并运行

@SpringBootConfiguration
@EnableAsync
@EnableScheduling
@ComponentScan({"org.thingsboard.server.mqtt", "org.thingsboard.server.common", "org.thingsboard.server.transport.mqtt", "org.thingsboard.server.kafka"})
public class ThingsboardMqttTransportApplication {

    private static final String SPRING_CONFIG_NAME_KEY = "--spring.config.name";
    private static final String DEFAULT_SPRING_CONFIG_PARAM = SPRING_CONFIG_NAME_KEY + "=" + "tb-mqtt-transport";

    public static void main(String[] args) {
        SpringApplication.run(ThingsboardMqttTransportApplication.class, updateArguments(args));
    }

    private static String[] updateArguments(String[] args) {
        if (Arrays.stream(args).noneMatch(arg -> arg.startsWith(SPRING_CONFIG_NAME_KEY))) {
            String[] modifiedArgs = new String[args.length + 1];
            System.arraycopy(args, 0, modifiedArgs, 0, args.length);
            modifiedArgs[args.length] = DEFAULT_SPRING_CONFIG_PARAM;
            return modifiedArgs;
        }
        return args;
    }
}

第2-3行代码中，@EnableAsync注解使用来开启异步线程，@EnableScheduling注解使用来开启定时任务。

第4行代码@ComponentScan({"org.thingsboard.server.mqtt", "org.thingsboard.server.common", "org.thingsboard.server.transport.mqtt", "org.thingsboard.server.kafka"}): 扫描这些包下的所有使用@Component 的类，不管自动导入还是导出。

第7-8行代码和updateArguments的作用是：启动时，使用 --spring.config.name = tb-mqtt-transport, 指定配置名，包括但不仅限于tb-mqtt-transport.conf等文件。

MQTT Transport Common

Netty框架

Thingsboard的Mqtt协议逻辑实现是通过Netty实现的，Netty是一个NIO客户端、服务器框架，可以快速轻松地开发网络应用程序，例如协议服务器和客户端。它极大地简化和简化了网络编程，例如TCP和UDP套接字服务器。

Netty 官网
Netty GitHub仓库

引入依赖

MQTT Transport common通过引入Netty 4.x版本的jar包对Mqtt进行协议逻辑实现，Netty4.x和3.x的区别还是挺大的。

<dependency>
	<groupId>io.netty</groupId>
	<artifactId>netty-all</artifactId>
</dependency>

参数配置

transport:
	 # 本地MQTT传输参数
  mqtt:
    # 开启/关闭mqtt传输协议.
    enabled: "${MQTT_ENABLED:true}"
    bind_address: "${MQTT_BIND_ADDRESS:0.0.0.0}"
    bind_port: "${MQTT_BIND_PORT:1883}"
    timeout: "${MQTT_TIMEOUT:10000}"
    netty:
      leak_detector_level: "${NETTY_LEAK_DETECTOR_LVL:DISABLED}"
      boss_group_thread_count: "${NETTY_BOSS_GROUP_THREADS:1}"
      worker_group_thread_count: "${NETTY_WORKER_GROUP_THREADS:12}"
      max_payload_size: "${NETTY_MAX_PAYLOAD_SIZE:65536}"
      so_keep_alive: "${NETTY_SO_KEEPALIVE:false}"
    # MQTT SSL配置
    ssl:
      # 开启/关闭 SSL支持
      enabled: "${MQTT_SSL_ENABLED:false}"
      # SSL协议: 参阅http://docs.oracle.com/javase/8/docs/technotes/guides/security/StandardNames.html#SSLContext
      protocol: "${MQTT_SSL_PROTOCOL:TLSv1.2}"
      # Path to the key store that holds the SSL certificate
      key_store: "${MQTT_SSL_KEY_STORE:mqttserver.jks}"
      # Password used to access the key store
      key_store_password: "${MQTT_SSL_KEY_STORE_PASSWORD:server_ks_password}"
      # Password used to access the key
      key_password: "${MQTT_SSL_KEY_PASSWORD:server_key_password}"
      # Type of the key store
      key_store_type: "${MQTT_SSL_KEY_STORE_TYPE:JKS}"

模块目录结构

首先我们看该模块下的目录结构:

└── main
    └── java
        └── org
            └── thingsboard
                └── server
                    └── transport
                        └── mqtt
                            ├── MqttSslHandlerProvider.java	Mqtt Ssl逻辑处理提供类
                            ├── MqttTopics.java	Mqtt预定义主题
                            ├── MqttTransportContext.java Mqtt传输协议上下文
                            ├── MqttTransportHandler.java Mqttt传输协议逻辑处理类
                            ├── MqttTransportServerInitializer.java Mqtt传输协议初始化类
                            ├── MqttTransportService.java	Mqtt传输协议启动类
                            ├── adaptors
                            │   ├── JsonMqttAdaptor.java Mqtt传输内容Json适配器
                            │   └── MqttTransportAdaptor.java Mqtt协议传输适配器
                            ├── session
                            │   ├── DeviceSessionCtx.java 设备会话上下文
                            │   ├── GatewayDeviceSessionCtx.java 网关设备会话上下文
                            │   ├── GatewaySessionHandler.java 网关会话处理类
                            │   ├── MqttDeviceAwareSessionContext.java Mqtt设备会话上下文
                            │   └── MqttTopicMatcher.java Mqtt主题匹配器
                            └── util
                                └── SslUtil.java Ssl工具类

Mqtt传输协议逻辑实现

MqttTransportService

@Service("MqttTransportService")
@ConditionalOnExpression("'${transport.type:null}'=='null' || ('${transport.type}'=='local' && '${transport.mqtt.enabled}'=='true')")
@Slf4j
public class MqttTransportService {

    @Value("${transport.mqtt.bind_address}")
    private String host;
    @Value("${transport.mqtt.bind_port}")
    private Integer port;

    @Value("${transport.mqtt.netty.leak_detector_level}")
    private String leakDetectorLevel;
    @Value("${transport.mqtt.netty.boss_group_thread_count}")
    private Integer bossGroupThreadCount;
    @Value("${transport.mqtt.netty.worker_group_thread_count}")
    private Integer workerGroupThreadCount;
    @Value("${transport.mqtt.netty.so_keep_alive}")
    private boolean keepAlive;

    @Autowired
    private MqttTransportContext context;

    private Channel serverChannel;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    @PostConstruct
    public void init() throws Exception {
        log.info("Setting resource leak detector level to {}", leakDetectorLevel);
        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.valueOf(leakDetectorLevel.toUpperCase()));

        log.info("Starting MQTT transport...");
        bossGroup = new NioEventLoopGroup(bossGroupThreadCount);
        workerGroup = new NioEventLoopGroup(workerGroupThreadCount);
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new MqttTransportServerInitializer(context))
                .childOption(ChannelOption.SO_KEEPALIVE, keepAlive);

        serverChannel = b.bind(host, port).sync().channel();
        log.info("Mqtt transport started!");
    }

    @PreDestroy
    public void shutdown() throws InterruptedException {
        log.info("Stopping MQTT transport!");
        try {
            serverChannel.close().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
        log.info("MQTT transport stopped!");
    }
}

第6行到18行, 通过@value来注入对应的值,直接在字段上添加@value获取application.yml文件中的值。

MQTT服务端参数:

物联网mqtt从搭建到实战教程,mqtt物联网使用协议

其中第三个内存检测级别是源于: netty中大量使用了池化技术来减缓IO buffer的创建销毁开销。对于这些内存池管理的对象，从netty 4之后使用了引用计数来对它们进行管理。但是JVM GC和netty的内存回收机制是不同的，netty就提供了一个内存泄漏检查机制。

DISABLED: 不进行内存泄露的检测；
SIMPLE: 抽样检测，且只对部分方法调用进行记录，消耗较小，有泄漏时可能会延迟报告，默认级别；
ADVANCED: 抽样检测，记录对象最近几次的调用记录，在泄漏时可能会延迟报告；
PARANOID: 每次创建一个对象时都进行泄露检测，且会记录对象最近的详细调用记录。是比较激进的内存泄露检测级别，消耗最大，建议只在测试时使用。

第四个和第五个参数,boss线程组用于服务端接受客户端的连接,worker线程组用于进行SocketChannel的数据读写。

第35行到第41行，通过创建ServerBootstrap对象,设置使用的EventLoopGroup;设置要被实例化的为NioServerSocketChannel类;设置连入服务端的Client的SocketChannel的处理器，绑定端口，并同步等待成功，即启动服务器。

第46行到56行，监听服务端关闭，并阻塞等待，并优雅关闭两个EventLoopGroup对象。

MqttTransportServerInitializer

public class MqttTransportServerInitializer extends ChannelInitializer<SocketChannel> {

    private final MqttTransportContext context;

    public MqttTransportServerInitializer(MqttTransportContext context) {
        this.context = context;
    }

    @Override
    public void initChannel(SocketChannel ch) {
        ChannelPipeline pipeline = ch.pipeline();
        if (context.getSslHandlerProvider() != null) {
            SslHandler sslHandler = context.getSslHandlerProvider().getSslHandler();
            pipeline.addLast(sslHandler);
            context.setSslHandler(sslHandler);
        }
        pipeline.addLast("decoder", new MqttDecoder(context.getMaxPayloadSize()));
        pipeline.addLast("encoder", MqttEncoder.INSTANCE);

        MqttTransportHandler handler = new MqttTransportHandler(context);

        pipeline.addLast(handler);
        ch.closeFuture().addListener(handler);
    }

}

第11-16行设置ChannelPipeLine，判断SSL处理器处理类是否为空，如果不为空，将SSL处理器加入到ChannelPipeLine。

第17-23行，添加负载内容的解编码器，Mqtt协议逻辑处理器和异步操作完成时回调。

MqttTransportHandler

物联网mqtt从搭建到实战教程,mqtt物联网使用协议

总结

让大家了解MQTT是什么？MQTT是事实上的物联网准标准协议，且经过多年的发展，在IoT行业已然长成参天大树。
基于NIO通讯的Netty框架是实现MQTT broker的不二良选。
带大家初步了解怎么用Netty框架搭建一个MQTT服务，Netty的网络参数又是怎么样！

物联网mqtt协议浅谈 (物联网mqtt协议内容)

源码解析系列

a.『 准备篇 』

b.『设备连接协议篇 』

MQTT概述

关联模块一览

MQTT Transport Service

MQTT Transport Common

模块目录结构

Mqtt传输协议逻辑实现

总结

a.『准备篇』

b.『设备连接协议篇』