脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → SpringBoot异步线程调用

SpringBoot整合MQTT并实现异步线程调用的问题

 更新时间：2021年11月24日 10:20:07   作者：时间静止不是简史  
这篇文章主要介绍了基于SpringBoot通过注解实现对mqtt消息处理的异步调用,本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下

为什么选择MQTT

MQTT的定义相信很多人都能讲的头头是道,本文章也不讨论什么高大上的东西,旨在用最简单直观的方式让每一位刚接触的同行们可以最快的应用起来

先从使用MQTT需要什么开始分析:

  • 消息服务器
  • 不同应用/设备之间的频繁交互
  • 可能涉及一对多的消息传递

基于SpringBoot通过注解实现对mqtt消息处理的异步调用

使用背景

生产环境下, 由于mqtt 生产者生产的消息逐渐增多, 可能会导致消息堆积. 因此需要消费者去快速的消费.
而其中的一个方案便是使用异步线程去加速消费消息. 下面介绍下思路

我们可以在原来的mqtt工具类上面进行改装.
首先创建一个类MqttMessageListener并继承IMqttMessageListener实现messageArrived, 用于处理这些消息(业务编写)
然后改写mqtt客户端订阅的方法, 注入MqttMessageListener, 并在订阅方法中新增该参数
在然后在启动类开启异步线程, 编写一个配置类配置线程池参数并且在messageArrived加上@Async开启异步线程调用

代码实现

基础代码

指没有开启线程池的代码

MqttPushClient 主要定义了连接参数

import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Author
 * @Date
 * @Description  连接至EMQ X 服务器，获取mqtt连接，发布消息
 */
@Component
public class MqttPushClient{

    private static final Logger log = LoggerFactory.getLogger(MqttPushClient.class);

    @Autowired
    private PushCallback pushCallback;

    private static MqttClient client;

    public static void setClient(MqttClient client) {
        MqttPushClient.client = client;
    }

    public static MqttClient getClient() {
        return client;
    }

    public void connect(String host, String clientID, String username, String password, int timeout, int keepalive, List<String> topicList) {
        MqttClient client;
        try {
            client = new MqttClient(host, clientID, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            if (username != null) {
                options.setUserName(username);
            }
            if (password != null) {
                options.setPassword(password.toCharArray());
            }
            options.setConnectionTimeout(timeout);
            options.setKeepAliveInterval(keepalive);
            MqttPushClient.setClient(client);
            try {
                //设置回调类
                client.setCallback(pushCallback);
                //client.connect(options);
                IMqttToken iMqttToken = client.connectWithResult(options);
                boolean complete = iMqttToken.isComplete();
                log.info("MQTT连接"+(complete?"成功":"失败"));
                /** 订阅主题 **/
                for (String topic : topicList) {
                    log.info("连接订阅主题：{}", topic);
                    client.subscribe(topic, 0);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

PushCallback 回调类, 实现重连, 消息发送监听, 消息接收监听

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @Author
 * @Date
 * @Description  消息回调，处理接收的消息
 */
@Component
public class PushCallback implements MqttCallback {

    private static final Logger log = LoggerFactory.getLogger(PushCallback.class);

    @Autowired
    private MqttConfiguration mqttConfiguration;
    @Autowired
    private MqttTopic mqttTopic;

    @Override
    public void connectionLost(Throwable cause) {        // 连接丢失后，一般在这里面进行重连
        log.info("连接断开，正在重连");
        MqttPushClient mqttPushClient = mqttConfiguration.getMqttPushClient();
        if (null != mqttPushClient) {
            mqttPushClient.connect(mqttConfiguration.getHost(), mqttConfiguration.getClientid(), mqttConfiguration.getUsername(),
                    mqttConfiguration.getPassword(), mqttConfiguration.getTimeout(), mqttConfiguration.getKeepalive(), mqttConfiguration.getTopic());
            log.info("已重连");
        }

    }

    /**
     * 发送消息，消息到达后处理方法
     * @param token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        int messageId = token.getMessageId();
        String[] topics = token.getTopics();
        log.info("消息发送完成,messageId={},topics={}",messageId,topics.toString());
    }

    /**
     * 订阅主题接收到消息处理方法
     * @param topic
     * @param message
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) {
         // subscribe后得到的消息会执行到这里面,这里在控制台有输出
        String messageStr = new String(message.getPayload());
        // messageDistribute.distribute(topic, messageStr);
        log.info("接收的主题：" + topic +  ";接收到的信息：" + messageStr);
        }
  }

MqttConfiguration 配置了mqtt相关参数, 并初始化连接(mqtt在这里启动)

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @Author
 * @Date mqtt配置及连接
 * @Description
 */
@Slf4j
@Component
@Configuration
@ConfigurationProperties(MqttConfiguration.PREFIX)
public class MqttConfiguration {

    @Autowired
    private MqttPushClient mqttPushClient;

    /**
     * 指定配置文件application-local.properties中的属性名前缀
     */
    public static final String PREFIX = "std.mqtt";

    private String host;
    private String clientId;
    private String userName;
    private String password;
    private int timeout;
    private int keepAlive;
    private List<String> topic;

    public String getClientid() {
        return clientId;
    }

    public void setClientid(String clientid) {
        this.clientId = clientid;
    }

    public String getUsername() {
        return userName;
    }

    public void setUsername(String username) {
        this.userName = username;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public int getTimeout() {
        return timeout;
    }

    public void setTimeout(int timeout) {
        this.timeout = timeout;
    }

    public int getKeepalive() {
        return keepAlive;
    }

    public void setKeepalive(int keepalive) {
        this.keepAlive = keepalive;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public List<String> getTopic() {
        return topic;
    }

    public void setTopic(List<String> topic) {
        this.topic = topic;
    }

    /**
     * 连接至mqtt服务器，获取mqtt连接
     * @return
     */
    @Bean
    public MqttPushClient getMqttPushClient() {
        //连接至mqtt服务器，获取mqtt连接
        mqttPushClient.connect(host, clientId, userName, password, timeout, keepAlive, topic);
        return mqttPushClient;
    }
}

properties.yml 配置文件
std.mqtt:
  host: tcp://x.x.x.x:1883
  username: your_username
  password: your_password
  #MQTT-连接服务器默认客户端ID
  clientid: your_clientid
  #连接超时
  timeout: 1000
  # deviceId
  deviceId: your_deviceId
  # mqtt-topic
  topic[0]: your_tpoic

TopicOperation 定义了发布订阅的方法

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * @Author chy
 */
public class TopicOperation {

    private static final Logger log = LoggerFactory.getLogger(TopicOperation.class);

    /**
     * 订阅主题
     * @param topic 主题名称
     */
    public static void subscribe(String topic) {
        try {
            MqttClient client = MqttPushClient.getClient();
            if (client == null) {
                return;
            };
            client.subscribe(topic, 0);
            log.info("订阅主题:{}",topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布主题
     *
     * @param topic
     * @param pushMessage
     */
    public static void publish(String topic, String pushMessage) {

        log.info("SEND TO MQTT -- topic : {}, message : {}", topic, pushMessage);

        MqttMessage message = new MqttMessage();
        message.setQos(0);
        // 非持久化
        message.setRetained(false);
        message.setPayload(pushMessage.getBytes());
        MqttClient client = MqttPushClient.getClient();
        if (client == null) {
            return;
        };
        MqttTopic mTopic = client.getTopic(topic);
        if (null == mTopic) {
            log.error("主题不存在:{}",mTopic);
        }
        try {
            mTopic.publish(message);
        } catch (Exception e) {
            log.error("mqtt发送消息异常:",e);
        }
    }

}

定义了发布和订阅的相关主题

import com.sxd.onlinereservation.exception.BusinessException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * @Author
 * @Date topic名称
 * @Description
 */
@Component
public class MqttTopic {

    @Value("${std.mqtt.deviceId}")
    private String[] deviceId;

    public String getSubscribeTopic(String type){
        switch (type){

            case "appointTopic":
                return String.format("/v1/%s/service/appointTopic", deviceId[0]);
            default:
                throw new BusinessException("mqtt 订阅主题获取错误");
        }
    }

    public String getPublishTopic(String type) {
        switch (type){
                //1.0接口立即取号发布主题
            case "appointTopic":
                return String.format("/v1/%s/service/appointTopic", deviceId[1]);
            default:
                throw new BusinessException("mqtt 发布主题获取错误");
        }
    }
}

ps: 如果想要使用该工具类进行消息发送和接收看下面demo

//消息发布操作
 TopicOperation.publish(mqttTopic.getPublishTopic("appointTopic"), "消息体"));
 //消息订阅操作
  TopicOperation.subscribe(mqttTopic.getSubscribeTopic("appointTopic"), "消息体"));

异步线程处理实现

总结

  • 创建消息监听类 , 用于监听消息并进行业务处理
  • 在原来订阅时, 注入并使用第一步创建的监听类
  • 通过注解开启异步线程并配置处理方式

创建消息监听类 , 用于监听消息并进行业务处理

@Slf4j
@Component
public class MqttMessageListener implements IMqttMessageListener {

    @Resource
    private BusinessService businessService;
    @Autowired
    private MqttTopic mqttTopic;
    @Autowired
    private ThreeCallmachineService threeCallmachineService;
    @Autowired
    private BusinessHallService businessHallService;
    @Autowired
    private BusinessMaterialService businessMaterialService;
    @Autowired
    private BusinessWaitService businessWaitService;
    @Autowired
    private AppointmentService appointmentService;


    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String messageStr = new String(message.getPayload());
        log.info("接收的主题：" + topic +  ";接收到的信息：" + messageStr);
        //进行 业务处理
        }
}

在原来订阅时, 注入并使用第一步创建的监听类

注入了 MqttMessageListener , 并且在订阅时加入 client.subscribe(topic, mqttMessageListener);

修改MqttPushClient (必须)

@Component
public class MqttPushClient{

    private static final Logger log = LoggerFactory.getLogger(MqttPushClient.class);

    @Autowired
    private PushCallback pushCallback;
    @Autowired   //这里进行了注入操作
    private MqttMessageListener mqttMessageListener;

    private static MqttClient client;

    public static void setClient(MqttClient client) {
        MqttPushClient.client = client;
    }

    public static MqttClient getClient() {
        return client;
    }

    public void connect(String host, String clientID, String username, String password, int timeout, int keepalive, List<String> topicList) {
        MqttClient client;
        try {
            client = new MqttClient(host, clientID, new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            if (username != null) {
                options.setUserName(username);
            }
            if (password != null) {
                options.setPassword(password.toCharArray());
            }
            options.setConnectionTimeout(timeout);
            options.setKeepAliveInterval(keepalive);
            MqttPushClient.setClient(client);
            try {
                //设置回调类
                client.setCallback(pushCallback);
                //client.connect(options);
                IMqttToken iMqttToken = client.connectWithResult(options);
                boolean complete = iMqttToken.isComplete();
                log.info("MQTT连接"+(complete?"成功":"失败"));
                /** 订阅主题 **/
                for (String topic : topicList) {
                    log.info("连接订阅主题：{}", topic);
                    //client.subscribe(topic, 0);
                    client.subscribe(topic, mqttMessageListener);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

如果业务还使用了手动订阅, 则也需要在订阅的类上面注入MqttMessageListener , 并且在订阅方法中作为参数使用. 但是我们需要将方法改成非静态的, 因此在使用该方法时我们需要new该对象然后才能够调用. 但是手动订阅很少用到. 因此有无此步骤都可

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * @Author chy
 * @Date
 * @Description
 */
public class TopicOperation {

    private static final Logger log = LoggerFactory.getLogger(TopicOperation.class);

	//注入MqttMessageListener
    @Autowired
    private MqttMessageListener mqttMessageListener;

    /**
     * 订阅主题
     * @param topic 主题名称
     */
    public void subscribe(String topic) {
        try {
            MqttClient client = MqttPushClient.getClient();
            if (client == null) {
                return;
            };
           //client.subscribe(topic, 0);
           //在订阅方法中作为参数使用
            client.subscribe(topic, mqttMessageListener);
            log.info("订阅主题:{}",topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布主题
     *
     * @param topic
     * @param pushMessage
     */
    public static void publish(String topic, String pushMessage) {

        log.info("SEND TO MQTT -- topic : {}, message : {}", topic, pushMessage);

        MqttMessage message = new MqttMessage();
        message.setQos(0);
        // 非持久化
        message.setRetained(false);
        message.setPayload(pushMessage.getBytes());
        MqttClient client = MqttPushClient.getClient();
        if (client == null) {
            return;
        };
        MqttTopic mTopic = client.getTopic(topic);
        if (null == mTopic) {
            log.error("主题不存在:{}",mTopic);
        }
        try {
            mTopic.publish(message);
        } catch (Exception e) {
            log.error("mqtt发送消息异常:",e);
        }
    }

}

通过注解开启异步线程并配置处理方式 启动类开启 @EnableAsync(proxyTargetClass=true )

@SpringBootApplication
@MapperScan(basePackages = "com.x.x.mapper")
@EnableTransactionManagement
@EnableAsync(proxyTargetClass=true )
public class XXApplication {

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

}

配置类配置线程池参数

@Slf4j
@Configuration
public class ExecutorConfig {

    @Bean
    public Executor asyncServiceExecutor() {
        log.info("start asyncServiceExecutor");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(9);
        //配置最大线程数
        executor.setMaxPoolSize(20);
        //配置队列大小
        executor.setQueueCapacity(200);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix("sxd-async-service-");
        // 设置拒绝策略：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //执行初始化
        executor.initialize();
        return executor;
    }
}

MqttMessageListener的实现方法messageArrived开启@Async("asyncServiceExecutor")

@Slf4j
@Component
public class MqttMessageListener implements IMqttMessageListener {

    @Resource
    private BusinessService businessService;
    @Autowired
    private MqttTopic mqttTopic;
    @Autowired
    private ThreeCallmachineService threeCallmachineService;
    @Autowired
    private BusinessHallService businessHallService;
    @Autowired
    private BusinessMaterialService businessMaterialService;
    @Autowired
    private BusinessWaitService businessWaitService;
    @Autowired
    private AppointmentService appointmentService;


    @Override
    @Async("asyncServiceExecutor")
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String messageStr = new String(message.getPayload());
        log.info("接收的主题：" + topic +  ";接收到的信息：" + messageStr);
        System.out.println("线程名称：【" + Thread.currentThread().getName() + "】");
        //进行 业务处理
        }
}

到此这篇关于SpringBoot整合MQTT并实现异步线程调用的文章就介绍到这了,更多相关SpringBoot异步线程调用内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • Jenkins发送测试报告邮件过程详解

    Jenkins发送测试报告邮件过程详解

    这篇文章主要介绍了Jenkins发送测试报告邮件过程详解,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-07-07
  • Spring的初始化和XML解析的实现

    Spring的初始化和XML解析的实现

    这篇文章主要介绍了Spring的初始化和XML解析的实现，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
    2019-03-03
  • 浅析Java中的继承与组合

    浅析Java中的继承与组合

    本文将介绍组合和继承的概念及区别，并从多方面分析在写代码时如何进行选择。文中通过示例代码介绍的很详细,有需要的朋友可以参考借鉴，下面来一起看看吧。
    2016-12-12
  • Java数据结构之环形链表和约瑟夫问题详解

    Java数据结构之环形链表和约瑟夫问题详解

    约瑟夫(Josephus)问题是单向环形链表的一种体现,也就是丢手帕问题,下面这篇文章主要给大家介绍了关于Java数据结构之环形链表和约瑟夫问题的相关资料,文中通过实例代码介绍的非常详细,需要的朋友可以参考下
    2022-08-08
  • java如何动态的处理接口的返回数据

    java如何动态的处理接口的返回数据

    本文主要介绍了java如何动态的处理接口的返回数据，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2023-01-01
  • Java如何实现登录token令牌

    Java如何实现登录token令牌

    这篇文章主要介绍了Java如何实现登录token令牌，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2022-05-05
  • spring boot的健康检查HealthIndicators实战

    spring boot的健康检查HealthIndicators实战

    这篇文章主要介绍了spring boot的健康检查HealthIndicators实战，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-10-10
  • SpringBoot从繁至简的框架基础教程

    SpringBoot从繁至简的框架基础教程

    Spring Boot是由Pivotal团队提供的全新框架，其设计目的是用来简化新Spring应用的初始搭建以及开发过程。该框架使用了特定的方式来进行配置，从而使开发人员不再需要定义样板化的配置
    2022-10-10
  • SpringBoot生产环境和测试环境配置分离的教程详解

    SpringBoot生产环境和测试环境配置分离的教程详解

    这篇文章主要介绍了SpringBoot生产环境和测试环境配置分离的教程详解,需要的朋友可以参考下
    2020-08-08
  • Java中的对象、类、抽象类、接口、继承之间的联系

    Java中的对象、类、抽象类、接口、继承之间的联系

    这篇文章主要介绍了Java中的对象、类、抽象类、接口、继承之间的联系，文章讲解的很清晰，有不太懂的同学可以多研究下
    2021-02-02

最新评论