其他分享
首页 > 其他分享> > MQTT接收HEX(2/2)

MQTT接收HEX(2/2)

作者:互联网

一些通讯硬件默认发送和接收的是UTF-8字符和ASCII的消息,但也是有一些通讯硬件的是通过16进制消息进行交互的。

一、配置pom的maven依赖

<dependency>
    <groupId>org.fusesource.hawtbuf</groupId>
    <artifactId>hawtbuf</artifactId>
    <version>1.11</version>
</dependency>
<dependency>
    <groupId>org.fusesource.hawtdispatch</groupId>
    <artifactId>hawtdispatch</artifactId>
    <version>1.22</version>
</dependency>
<dependency>
    <groupId>org.fusesource.hawtdispatch</groupId>
    <artifactId>hawtdispatch-transport</artifactId>
    <version>1.22</version>
</dependency>
<dependency>
    <groupId>org.fusesource.mqtt-client</groupId>
    <artifactId>mqtt-client</artifactId>
    <version>1.16</version>
</dependency>

 

为了代码的可阅读性,我将以下代码按照功能封装到不同的类当中。

 

二、MQTT数据接收类

package com.xxxx.worker.controller.Server.MQTT;

import com.xxxx.common.utils.PropertyUtils;
import com.xxxx.worker.controller.Server.Execute.MessageExecute;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;

/**
 * @title: MQTT消息订阅(接收十六进制)
 * @author: hunttown
 * @date: 2021年01月06日 16:32
 * @description: 当前类用bean注入,不要使用注解注入,因为下面要将其配置到worker里。
 */
public class MQTTSubHex {

    //服务器地址
    private final static String serverUrl = "你的服务器IP:端口";

    //客户端唯一标识
    private final static String clientid = "随便起个名字";

    //订阅主题
    private final static String subtopic = "你的订阅主题";

    //用户名
    private final static String username = "用户名";

    //密码
    private final static String password = "密码";

    //传输质量:0至多一次;1至少一次;2确保只有一次。
    private final static int qos = 0;

    void start() {
        try {
            //创建MQTT对象
            MQTT mqtt = new MQTT();

            // 设置MQTT Broker的ip和端口
            mqtt.setHost(serverUrl);

            // 连接前清空会话信息
            mqtt.setCleanSession(true);

            // 设置重新连接的次数
            mqtt.setReconnectAttemptsMax(10);

            // 设置重连的间隔时间(毫秒)
            mqtt.setReconnectDelay(2000);

            // 设置心跳时间(秒)
            mqtt.setKeepAlive((short) 30);

            // 设置缓冲的大小
            mqtt.setSendBufferSize(2 * 1024 * 1024);

            // 设置客户端ID
            mqtt.setClientId(clientid);

            // 设置用户名和字码
            mqtt.setUserName(username);
            mqtt.setPassword(password);

            final FutureConnection connection = mqtt.futureConnection();
            connection.connect();

            Topic[] topics = {new Topic(subtopic, QoS.AT_LEAST_ONCE)};
            connection.subscribe(topics);

            System.out.println("MQTT订阅设置初始化完毕!");

            MessageExecute hexExecute = new MessageExecute();

            while (true) {
                //接收信息
                Future<Message> futrueMessage = connection.receive();
                Message message = futrueMessage.await();

                String msg = String.valueOf(message.getPayloadBuffer());
                System.out.println("接收到信息:" + msg);

                //这里开始处理你的业务
                //1、如果数据量较大,可使用中间件暂存信息,如:MQ
                //2、如果数据量较小,可以使用异步处理
                //TODO
            }

        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("MQTT订阅设置初始化失败!");
        }
    }
}

 

三、线程类

package com.xxxx.worker.controller.Server.MQTT;

/**
 * @title: 启动一个线程
 * @author: hunttown
 * @date: 2021年03月23日 15:55
 * @description:
 */
public class MQTTHexThread extends Thread {

    private static MQTTSubHex mqttSubHex;

    public void run() {
        mqttSubHex.start();
    }

    public static MQTTSubHex getMqttSubHex() {
        return mqttSubHex;
    }

    public static void setMqttSubHex(MQTTSubHex mqttSubHex) {
        MQTTHexThread.mqttSubHex = mqttSubHex;
    }
}

 

四、监听类

package com.xxxx.worker.controller.TaskListener;

import com.xxxx.worker.controller.Server.MQTT.MQTTHexThread;

import javax.servlet.ServletContextEvent;
import javax.servlet.ServletContextListener;
import javax.servlet.http.HttpServlet;

/**
 * @title: MQTT监听类
 * @author: hunttown
 * @date: 2020年10月19日 18:06
 * @description: MQTT类
 */
public class MQTTHexListener extends HttpServlet implements ServletContextListener {
    public void contextInitialized(ServletContextEvent arg0) {

        System.out.println("----------------- MQTT:Hex线程已启动 --------------------------");

        MQTTHexThread thread = new MQTTHexThread();
        thread.setDaemon(true); //设置线程为后台线程
        thread.start();
    }

    public void contextDestroyed(ServletContextEvent arg0) {
        // TODO
    }
}

 

五、将监听类配置到web.xml中

<listener>
    <!-- MQTT:接收Hex数据 -->
    <listener-class>com.hunttown.analysis.worker.controller.TaskListener.MQTTHexListener</listener-class>
</listener>

 

完毕!

 

标签:fusesource,HEX,mqtt,MQTT,static,org,import,接收
来源: https://www.cnblogs.com/hunttown/p/16188905.html