Springboot整合mqtt服务的示例代码

更新时间：2022年03月18日 15:56:52 作者：李泰山

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。本文为大家分享了Springboot整合mqtt服务的示例代码，需要的可以参考一下

首先在pom文件里引入mqtt的依赖配置

        <!--mqtt-->
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.4</version>
        </dependency>

其次在springboot 的配置yml文件，配置mqtt的服务配置

spring:  
  mqtt:
    url: tcp://127.0.0.1:1883
    client-id: niubility-tiger
    username:
    password:
    topic: [/unify/test]

创建 MqttProperties配置参数类

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
 
@Data
@ConfigurationProperties("spring.mqtt")
public class MqttProperties {
    private String url;
    private String clientId;
    private String username;
    private String password;
    private String[] topic;
}

创建 MqttConfiguration 配置类

import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springblade.core.tool.utils.Func;
import org.springblade.ubw.listener.MqttSubscribeListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
@EnableConfigurationProperties({MqttProperties.class})
public class MqttConfiguration {
    private static final Logger log = LoggerFactory.getLogger(MqttConfiguration.class);
    @Autowired
    private MqttProperties mqttProperties;
 
    public MqttConfiguration() {
    }
 
    @Bean
    public MqttConnectOptions mqttConnectOptions() {
        MqttConnectOptions connectOptions = new MqttConnectOptions();
        connectOptions.setServerURIs(new String[]{this.mqttProperties.getUrl()});
        if (Func.isNotBlank(this.mqttProperties.getUrl())) {
            connectOptions.setUserName(this.mqttProperties.getUsername());
        }
 
        if (Func.isNotBlank(this.mqttProperties.getPassword())) {
            connectOptions.setPassword(this.mqttProperties.getPassword().toCharArray());
        }
 
        connectOptions.setKeepAliveInterval(60);
        return connectOptions;
    }
 
    @Bean
    public IMqttClient mqttClient(MqttConnectOptions options) throws MqttException {
        IMqttClient mqttClient = new MqttClient(this.mqttProperties.getUrl(), this.mqttProperties.getClientId());
        mqttClient.connect(options);
        for(int i = 0; i< this.mqttProperties.getTopic().length; ++i) {
            mqttClient.subscribe(this.mqttProperties.getTopic()[i], new MqttSubscribeListener());
        }
        return mqttClient;
    }
}

创建订阅事件类

import org.springframework.context.ApplicationEvent;
 
 
public class UWBMqttSubscribeEvent extends ApplicationEvent {
    private String topic;
 
    public UWBMqttSubscribeEvent(String topic, Object source) {
        super(source);
        this.topic = topic;
    }
 
    public String getTopic() {
        return this.topic;
    }
}

创建订阅事件监听器

import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springblade.core.tool.utils.SpringUtil;
import org.springblade.ubw.event.UWBMqttSubscribeEvent;
 
 
public class MqttSubscribeListener implements IMqttMessageListener {
 
    @Override
    public void messageArrived(String s, MqttMessage mqttMessage) {
        String content = new String(mqttMessage.getPayload());
        UWBMqttSubscribeEvent event = new UWBMqttSubscribeEvent(s, content);
        SpringUtil.publishEvent(event);
    }
}

创建mqtt消息事件异步处理监听器

import com.baomidou.mybatisplus.core.toolkit.StringPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springblade.core.tool.utils.Func;
import org.springblade.ubw.config.MqttProperties;
import org.springblade.ubw.event.UWBMqttSubscribeEvent;
import org.springblade.ubw.service.MqttService;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
 
import javax.annotation.Resource;
import java.util.Arrays;
import java.util.List;
 
 
@Configuration
public class MqttEventListener {
 
 
    private static final Logger log = LoggerFactory.getLogger(MqttEventListener.class);
 
    @Resource
    private MqttProperties mqttProperties;
 
    @Resource
    private MqttService mqttService;
 
    private String processTopic (String topic) {
        List<String> topics = Arrays.asList(mqttProperties.getTopic());
        for (String wild : topics) {
            wild = wild.replace(StringPool.HASH, StringPool.EMPTY);
            if (topic.startsWith(wild)) {
                return topic.replace(wild, StringPool.EMPTY);
            }
        }
        return StringPool.EMPTY;
    }
 
 
    @Async
    @EventListener(UWBMqttSubscribeEvent.class)
    public void listen (UWBMqttSubscribeEvent event) {
        String topic = processTopic(event.getTopic());
        Object source = event.getSource();
        if (Func.isEmpty(source)) {
            return;
        }
        mqttService.issue(topic,source);
//        log.info("mqtt接收到 通道 {} 的信息为：{}",topic,source);
    }
}

创建MqttService 数据处理服务类

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springblade.core.tool.utils.Func;
import org.springblade.ubw.area.entity.WorkArea;
import org.springblade.ubw.area.entity.WorkSite;
import org.springblade.ubw.area.entity.WorkSiteNeighbourInfo;
import org.springblade.ubw.area.entity.WorkSitePassInfo;
import org.springblade.ubw.area.service.WorkAreaService;
import org.springblade.ubw.area.service.WorkSiteNeighbourInfoService;
import org.springblade.ubw.area.service.WorkSitePassInfoService;
import org.springblade.ubw.area.service.WorkSiteService;
import org.springblade.ubw.constant.UbwConstant;
import org.springblade.ubw.history.entity.HistoryLocusInfo;
import org.springblade.ubw.history.entity.HistoryOverTimeSosAlarmInfo;
import org.springblade.ubw.history.service.HistoryLocusInfoService;
import org.springblade.ubw.history.service.HistoryOverTimeSosAlarmInfoService;
import org.springblade.ubw.loc.entity.LocStatusInfo;
import org.springblade.ubw.loc.entity.LocStatusInfoHistory;
import org.springblade.ubw.loc.service.LocStatusInfoHistoryService;
import org.springblade.ubw.loc.service.LocStatusInfoService;
import org.springblade.ubw.msg.entity.*;
import org.springblade.ubw.msg.service.*;
import org.springblade.ubw.system.entity.*;
import org.springblade.ubw.system.service.*;
import org.springblade.ubw.system.wrapper.MqttWrapper;
import org.springframework.stereotype.Service;
 
import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;
 
 
@Service
public class MqttService {
 
    private static final Logger log = LoggerFactory.getLogger(MqttService.class);
 
    @Resource
    private EmployeeAndDepartmentService employeeAndDepartmentService;
 
    @Resource
    private VehicleInfoService vehicleInfoService;
 
    @Resource
    private WorkSiteService workSiteService;
 
    @Resource
    private LocStatusInfoService locStatusInfoService;
 
    @Resource
    private LocStatusInfoHistoryService locStatusInfoHistoryService;
 
    @Resource
    private LocOverTimeSosAlarminfoService locOverTimeSosAlarminfoService;
 
    @Resource
    private LocAreaOverSosAlarminfoService locAreaOverSosAlarmInfoService;
 
    @Resource
    private LocSosAlarminfoService locSosAlarminfoService;
 
    @Resource
    private AttendanceInfoService attendanceInfoService;
 
    @Resource
    private HistoryLocusInfoService historyLocusInfoService;
 
    @Resource
    private WorkSitePassInfoService workSitePassInfoService;
 
    @Resource
    private EnvironmentalMonitorInfoService environmentalMonitorInfoService;
 
    @Resource
    private TrAlertService trAlertService;
 
    @Resource
    private AddEvacuateInfoService addEvacuateInfoService;
 
    @Resource
    private CancelEvacuateInfoService cancelEvacuateInfoService;
 
    @Resource
    private WorkSiteNeighbourInfoService workSiteNeighbourInfoService;
 
    @Resource
    private LinkMsgAlarmInfoService linkMsgAlarmInfoService;
 
    @Resource
    private LeaderEmployeeInfoService leaderEmployeeInfoService;
 
    @Resource
    private ElectricMsgInfoService electricMsgInfoService;
 
    @Resource
    private WorkAreaService workAreaService;
 
    @Resource
    private HistoryOverTimeSosAlarmInfoService historyOverTimeSosAlarmInfoService;
 
    @Resource
    private SpecialWorksService specialWorksService;
 
    @Resource
    private AttendanceLocusInfoService attendanceLocusInfoService;
 
    @Resource
    private WorkTypeService workTypeService;
 
    @Resource
    private OfficePositionService officePositionService;
 
    @Resource
    private ClassTeamService classTeamService;
 
    /**
     * 方法描述: 消息分发
     *
     * @param topic
     * @param source
     * @author liwenbin
     * @date 2021年12月14日 14:14:09
     */
    public void issue(String topic,Object source){
        switch(topic){
            case UbwConstant.TOPIC_EMP :
                //人员和部门信息
                employeeAndDepartmentService.saveBatch(source);
                break;
            case UbwConstant.TOPIC_VEHICLE :
                //车辆信息
                List<VehicleInfo> vehicleInfos = MqttWrapper.build().toEntityList(source,new VehicleInfo());
                vehicleInfoService.deleteAll();
                vehicleInfoService.saveBatch(vehicleInfos);
                break;
            case UbwConstant.TOPIC_WORK_SITE :
                //基站信息
                List<WorkSite> workSites = MqttWrapper.build().toEntityList(source,new WorkSite());
                workSiteService.deleteAll();
                workSiteService.saveBatch(workSites);
                break;
            case UbwConstant.TOPIC_LOC_STATUS:
                //井下车辆人员实时
                List<LocStatusInfo> locStatusInfos = MqttWrapper.build().toEntityList(source,new LocStatusInfo());
                if (Func.isEmpty(locStatusInfos)){
                    break;
                }
                locStatusInfoService.deleteAll();
                //筛选入井人员列表
                List<LocStatusInfo> inWellList = locStatusInfos.stream().filter(s -> s.getIsInWell() == 1).collect(Collectors.toList());
                locStatusInfoService.saveBatch(inWellList);
                //人员历史数据入库
                List<LocStatusInfoHistory> locStatusInfoHistorys = MqttWrapper.build().toEntityList(source,new LocStatusInfoHistory());
                locStatusInfoHistoryService.saveBatch(locStatusInfoHistorys);
                break;
            case UbwConstant.TOPIC_LOC_OVER_TIME:
                //超时报警信息
                List<LocOverTimeSosAlarminfo> locOverTimeSosAlarmInfos = MqttWrapper.build().toEntityList(source,new LocOverTimeSosAlarminfo());
                locOverTimeSosAlarminfoService.saveBatch(locOverTimeSosAlarmInfos);
                break;
            case UbwConstant.TOPIC_LOC_OVER_AREA:
                //超员报警信息
                List<LocAreaOverSosAlarminfo> locAreaOverSosAlarmInfos = MqttWrapper.build().toEntityList(source,new LocAreaOverSosAlarminfo());
                locAreaOverSosAlarmInfoService.saveBatch(locAreaOverSosAlarmInfos);
                break;
            case UbwConstant.TOPIC_LOC_SOS:
                //求救报警信息
                List<LocSosAlarminfo> locSosAlarmInfos = MqttWrapper.build().toEntityList(source,new LocSosAlarminfo());
                locSosAlarminfoService.saveBatch(locSosAlarmInfos);
                break;
            case UbwConstant.TOPIC_ATTEND:
                //考勤信息
                List<AttendanceInfo> attendanceInfos = MqttWrapper.build().toEntityList(source,new AttendanceInfo());
                attendanceInfoService.saveBatch(attendanceInfos);
                break;
            case UbwConstant.TOPIC_HISTORY_LOCUS:
                //精确轨迹信息
                List<HistoryLocusInfo> historyLocusInfos = MqttWrapper.build().toEntityList(source,new HistoryLocusInfo());
                historyLocusInfoService.saveBatch(historyLocusInfos);
                break;
            case UbwConstant.TOPIC_WORK_SITE_PASS:
                //基站经过信息
                List<WorkSitePassInfo> workSitePassInfos = MqttWrapper.build().toEntityList(source,new WorkSitePassInfo());
                workSitePassInfoService.saveBatch(workSitePassInfos);
                break;
            case UbwConstant.TOPIC_ENV_MON:
                //环境监测信息
                List<EnvironmentalMonitorInfo> environmentalMonitorInfos = MqttWrapper.build().toEntityList(source,new EnvironmentalMonitorInfo());
                environmentalMonitorInfoService.saveBatch(environmentalMonitorInfos);
                break;
            case UbwConstant.TOPIC_TR_ALERT:
                //环境监测报警信息
                List<TrAlert> trAlerts = MqttWrapper.build().toEntityList(source,new TrAlert());
                trAlertService.saveBatch(trAlerts);
                break;
            case UbwConstant.TOPIC_ADD_EVA:
                //下发撤离信息
                List<AddEvacuateInfo> addEvacuateInfos = MqttWrapper.build().toEntityList(source,new AddEvacuateInfo());
                addEvacuateInfoService.saveBatch(addEvacuateInfos);
                break;
            case UbwConstant.TOPIC_CANCEL_EVA:
                //取消撤离信息
                List<CancelEvacuateInfo> cancelEvacuateInfos = MqttWrapper.build().toEntityList(source,new CancelEvacuateInfo());
                cancelEvacuateInfoService.saveBatch(cancelEvacuateInfos);
                break;
            case UbwConstant.TOPIC_WORK_SITE_NEI:
                //相邻基站关系信息
                workSiteNeighbourInfoService.deleteAll();
                List<WorkSiteNeighbourInfo> workSiteNeighbourInfos = MqttWrapper.build().toEntityList(source,new WorkSiteNeighbourInfo());
                workSiteNeighbourInfoService.saveBatch(workSiteNeighbourInfos);
                break;
            case UbwConstant.TOPIC_LINK_MSG:
                //基站链路信息
                linkMsgAlarmInfoService.deleteAll();
                List<LinkMsgAlarmInfo> linkMsgAlarmInfos = MqttWrapper.build().toEntityList(source,new LinkMsgAlarmInfo());
                linkMsgAlarmInfoService.saveBatch(linkMsgAlarmInfos);
                break;
            case UbwConstant.TOPIC_LEADER_EMP:
                //带班领导信息
                leaderEmployeeInfoService.deleteAll();
                List<LeaderEmployeeInfo> leaderEmployeeInfos = MqttWrapper.build().toEntityList(source,new LeaderEmployeeInfo());
                leaderEmployeeInfoService.saveBatch(leaderEmployeeInfos);
                break;
            case UbwConstant.TOPIC_ELE_MSG:
                //低电报警信息
                List<ElectricMsgInfo> electricMsgInfos = MqttWrapper.build().toEntityList(source,new ElectricMsgInfo());
                electricMsgInfoService.saveBatch(electricMsgInfos);
                break;
            case UbwConstant.TOPIC_WORK_AREA:
                //区域信息
                workAreaService.deleteAll();
                List<WorkArea> workAreas = MqttWrapper.build().toEntityList(source,new WorkArea());
                workAreaService.saveBatch(workAreas);
                break;
            case UbwConstant.TOPIC_HIS_OVER_TIME_SOS:
                //历史超时报警信息
                List<HistoryOverTimeSosAlarmInfo> historyOverTimeSosAlarmInfos = MqttWrapper.build().toEntityList(source,new HistoryOverTimeSosAlarmInfo());
                historyOverTimeSosAlarmInfoService.saveBatch(historyOverTimeSosAlarmInfos);
                break;
            case UbwConstant.TOPIC_SPECIAL_WORK:
                //特种人员预设线路信息
                specialWorksService.deleteAll();
                List<SpecialWorks> specialWorks = MqttWrapper.build().toEntityList(source,new SpecialWorks());
                specialWorksService.saveBatch(specialWorks);
                break;
            case UbwConstant.TOPIC_ATTEND_LOC:
                //历史考勤轨迹信息
                List<AttendanceLocusInfo> attendanceLocusInfos = MqttWrapper.build().toEntityList(source,new AttendanceLocusInfo());
                attendanceLocusInfoService.saveBatch(attendanceLocusInfos);
                break;
            case UbwConstant.TOPIC_WORK_TYPE:
                //工种信息
                workTypeService.deleteAll();
                List<WorkType> workTypes = MqttWrapper.build().toEntityList(source,new WorkType());
                workTypeService.saveBatch(workTypes);
                break;
            case UbwConstant.TOPIC_OFFICE_POS:
                //职务信息
                officePositionService.deleteAll();
                List<OfficePosition> officePositions = MqttWrapper.build().toEntityList(source,new OfficePosition());
                officePositionService.saveBatch(officePositions);
                break;
            case UbwConstant.TOPIC_CLASS_TEAM:
                //班组信息
                classTeamService.deleteAll();
                List<ClassTeam> classTeams = MqttWrapper.build().toEntityList(source,new ClassTeam());
                classTeamService.saveBatch(classTeams);
                break;
            default : //可选
                break;
        }
    }
}

完结，小伙伴们，可以根据这个demo 改造自己的mqtt服务处理！！！

以上就是Springboot整合mqtt服务的示例代码的详细内容，更多关于Springboot整合mqtt的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

Spring Boot+Vue实现Socket通知推送的完整步骤
最近工作中涉及消息通知功能的开发,所以下面这篇文章主要给大家介绍了关于Spring Boot+Vue实现Socket通知推送的完整步骤,文中通过实例代码以及图文介绍的非常详细,需要的朋友可以参考下
2023-04-04
Spring中自定义拦截器的使用
大家好，本篇文章主要讲的是Spring中自定义拦截器的使用，感兴趣的同学赶快来看一看吧，对你有帮助的话记得收藏一下
2022-02-02
使用JavaConfig代替xml实现Spring配置操作
这篇文章主要介绍了使用JavaConfig代替xml实现Spring配置操作，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-09-09
关于在Java中如何使用yaml的实例
这篇文章主要介绍了关于在Java中如何使用yaml的实例,YAML是一种轻量级的数据序列化格式。它以易读、易写的文本格式表示数据，支持列表、字典等各种数据结构，被广泛应用于配置文件、数据传输协议等领域,需要的朋友可以参考下
2023-08-08
详解多线程及Runable 和Thread的区别
这篇文章主要介绍了多线程及Runable 和Thread的区别，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2019-04-04
关于自定义过滤器获取不到session问题
这篇文章主要介绍了关于自定义过滤器获取不到session问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2022-01-01
你知道在Java中Integer和int的这些区别吗?
最近面试,突然被问道,说一下Integer和int的区别.额…可能平时就知道写一些业务代码,包括面试的一些Spring源码等,对于这种特别基础的反而忽略了,导致面试的时候突然被问到反而不知道怎么回答了.哎,还是乖乖再看看底层基础,顺带记录一下把 ,需要的朋友可以参考下
2021-06-06
java面试常见模式问题---单例模式
单例模式（Singleton Pattern）是 Java 中最简单的设计模式之一。这种类型的设计模式属于创建型模式，它提供了一种创建对象的最佳方式
2021-06-06
Spring Boot自动配置源码实例解析
Spring Boot作为Java领域最为流行的快速开发框架之一,其核心特性之一就是其强大的自动配置机制,下面这篇文章主要给大家介绍了关于Spring Boot自动配置源码的相关资料,文中通过代码介绍的非常详细,需要的朋友可以参考下
2024-08-08
详解Http协议以及post与get区别
这篇文章主要介绍了详解Http协议以及post与get区别,通过分别说明Http协议以及get与post各自的概念,再到两者作比较有着详细的说明,希望对你有所帮助
2021-06-06

Springboot整合mqtt服务的示例代码

相关文章

最新评论

大家感兴趣的内容

最近更新的内容

常用在线小工具