基于Netty实现的MQTT客户端详细文档+全部资料.zip

# netty-mqtt-client ## 1. 介绍 ### 1.1 基本概况 该项目是基于Netty实现的MQTT3及MQTT5协议的客户端，创建目的是为了学习和使用MQTT及Netty ### 1.2 技术栈 Java + Netty + MQTT ### 1.3 特色 1.基于高性能的网络开发框架Netty实现，性能更高 2.支持多个客户端使用同一个线程组，支持配置线程数量，占用的资源更少 3.目前支持MQTT 3.1.1以及MQTT 5版本 4.支持单向及双向SSL认证 5.支持自定义实现扩展组件 6.支持组件拦截，可实现插件扩展 7.代码全中文注释 8.支持消息持久化（目前支持Redis、内存、文件），仅保存不清理会话且未完成的客户端消息 9.支持遗嘱消息 10.支持QoS等级为：0、1、2 11.支持MQTT 3.1.1版本和MQTT 5版本相互切换，并且相互兼容 12.支持设置客户端的TCP连接参数 ### 1.4 组件介绍 #### MqttConfiguration ​ MQTT全局配置组件，可支持配置TCP连接参数，代理工厂，拦截器，IO线程数，组件创建器及消息存储器 #### MqttClientFactory ​ MQTT客户端工厂，用于创建客户端，只需要传递连接参数，即可根据全局配置创建对应的MQTT客户端 #### MqttMsgStore ​ MQTT消息存储器，默认是用内存消息存储器，如果需要持久化，可使用Redis或文件消息存储器 #### MqttClient ​ MQTT客户端，面向开发者的接口，包含所有的客户端操作API #### MqttConnectParameter ​ MQTT连接参数，包含MQTT3及MQTT5参数组合，通过设置不同的参数，可创建不同的客户端 #### MqttCallback ​ MQTT回调器，包含MQTT客户端中的所有回调，如消息发送完成回调、消息发送成功回调、连接相关回调、心跳回调、订阅回调等 #### MqttRetrier ​ MQTT重试器，用于重试QoS1及QoS2中失败或未完成的消息，可通过连接配置修改重试时间及间隔 #### MqttDelegateHandler ​ MQTT消息委托器，即MQTT客户端和Netty之间的桥梁，主要是把MQTT的消息和Netty之间的消息进行转换处理 #### MqttConnector ​ MQTT连接器，用于连接MQTT Broker，只负责连接工作 #### MqttChannelHandler ​ MQTT客户端在Netty中的出入栈的处理器，同时负责开启心跳的定时任务 #### MqttMsgIdCache ​ MQTT消息ID缓存器，用于生成MQTT协议层消息的ID #### ObjectCreator ​ 对象创建器，用于创建MqttClient、MqttConnector、MqttDelegateHandler三大组件，可自定义实现三大组件的创建及替换 #### ProxyFactory ​ 代理工厂，主要是用于拦截器，支持多种实现，目前支持JDK动态代理以及Cglib动态代理，默认使用JDK动态代理 #### Interceptor ​ 拦截器，仅支持拦截MqttClient、MqttConnector、MqttDelegateHandler三大组件，通过注解的方式使用，支持多层级拦截 ## 2.使用 ### 2.1 依赖 ``` <dependency> <groupId>io.github.xzc-coder</groupId> <artifactId>netty-mqtt-client</artifactId> <version>1.1.0</version> </dependency> ``` ### 2.2 初始化 ``` //创建MQTT全局配置器（也可以不创建） MqttConfiguration mqttConfiguration = new MqttConfiguration(2); //创建MQTT客户端工厂 MqttClientFactory mqttClientFactory = new DefaultMqttClientFactory(mqttConfiguration); //使用内存消息存储器（默认） MqttMsgStore mqttMsgStore = new MemoryMqttMsgStore(); mqttClientFactory.setMqttMsgStore(mqttMsgStore); //创建连接参数，设置客户端ID MqttConnectParameter mqttConnectParameter = new MqttConnectParameter("netty-mqtt-client-test"); //创建一个客户端 MqttClient mqttClient = mqttClientFactory.createMqttClient(mqttConnectParameter); ``` ### 2.3 连接 #### 连接参数设置 ##### MQTT 3 ``` //创建连接参数，设置客户端ID MqttConnectParameter mqttConnectParameter = new MqttConnectParameter("xzc_test"); //设置客户端版本（默认为3.1.1） mqttConnectParameter.setMqttVersion(MqttVersion.MQTT_3_1_1); //是否自动重连 mqttConnectParameter.setAutoReconnect(true); //Host mqttConnectParameter.setHost("broker.emqx.io"); //端口 mqttConnectParameter.setPort(1883); //是否使用SSL/TLS mqttConnectParameter.setSsl(false); //遗嘱消息 MqttWillMsg mqttWillMsg = new MqttWillMsg("test",new byte[]{},MqttQoS.EXACTLY_ONCE); mqttConnectParameter.setWillMsg(mqttWillMsg); //是否清除会话 mqttConnectParameter.setCleanSession(true); //心跳间隔 mqttConnectParameter.setKeepAliveTimeSeconds(60); //连接超时时间 mqttConnectParameter.setConnectTimeoutSeconds(30); //创建一个客户端 MqttClient mqttClient = mqttClientFactory.createMqttClient(mqttConnectParameter); //添加回调器 mqttClient.addMqttCallback(new DefaultMqttCallback()); ``` ##### MQTT 5 ``` //创建连接参数，设置客户端ID MqttConnectParameter mqttConnectParameter = new MqttConnectParameter("xzc_test"); //设置客户端版本 mqttConnectParameter.setMqttVersion(MqttVersion.MQTT_5_0_0); //是否自动重连 mqttConnectParameter.setAutoReconnect(true); //Host mqttConnectParameter.setHost("broker.emqx.io"); //端口 mqttConnectParameter.setPort(1883); //是否使用SSL/TLS mqttConnectParameter.setSsl(false); //遗嘱消息 MqttWillMsg mqttWillMsg = new MqttWillMsg("test",new byte[]{},MqttQoS.EXACTLY_ONCE); //MQTT 5的遗嘱属性 mqttWillMsg.setResponseTopic("test-response"); mqttWillMsg.setContentType("application/text"); mqttWillMsg.addMqttUserProperty("name","test"); mqttConnectParameter.setWillMsg(mqttWillMsg); //是否清除会话 mqttConnectParameter.setCleanSession(true); //心跳间隔 mqttConnectParameter.setKeepAliveTimeSeconds(60); //连接超时时间 mqttConnectParameter.setConnectTimeoutSeconds(30); //MQTT 5的连接参数 mqttConnectParameter.setMaximumPacketSize(100); mqttConnectParameter.setSessionExpiryIntervalSeconds(100); mqttConnectParameter.addMqttUserProperty("name","test"); //创建一个客户端 MqttClient mqttClient = mqttClientFactory.createMqttClient(mqttConnectParameter); //添加回调器 mqttClient.addMqttCallback(new DefaultMqttCallback()); ``` #### 连接API ``` /** * 进行连接，会阻塞至超时或者连接成功 */ void connect(); /** * 进行连接，不会阻塞 * * @return MqttFutureWrapper */ MqttFutureWrapper connectFuture(); ``` #### 示例 ``` //阻塞连接 mqttClient.connect(); ``` ``` //非阻塞连接 MqttFutureWrapper mqttFutureWrapper = mqttClient.connectFuture(); //添加监听器 mqttFutureWrapper.addListener(mqttFuture -> { if (mqttFuture.isSuccess()) { System.out.println("mqtt client connect success"); } else { System.out.println("mqtt client connect failure"); } }); ``` ### 2.4 断开连接 #### 断开连接API ``` /** * 断开连接，会阻塞至TCP断开 */ void disconnect(); /** * 断开连接 * * @return Future */ MqttFutureWrapper disconnectFuture(); /** * 断开连接（MQTT 5） * * @param mqttDisconnectMsg 断开消息 * @return Future */ MqttFutureWrapper disconnectFuture(MqttDisconnectMsg mqttDisconnectMsg); /** * 断开连接（MQTT 5） * * @param mqttDisconnectMsg 断开消息 */ void disconnect(MqttDisconnectMsg mqttDisconnectMsg); ``` #### 示例 ##### MQTT 3 ``` //阻塞断开连接 mqttClient.disconnect(); ``` ``` //非阻塞断开连接 MqttFutureWrapper mqttFutureWrapper = mqttClient.disconnectFuture(); //添加监听 mqttFutureWrapper.addListener(mqttFuture -> { if (mqttFuture.isDone()) { System.out.println("mqtt client disconnect done"); } }); ``` ##### MQTT 5 ``` //设置MQTT 5的断开连接参数 MqttDisconnectMsg mqttDisconnectMsg = new MqttDisconnectMsg(); mqttDisconnectMsg.setR