Springboot整合mqtt服务的示例代码

<!--mqtt-->

< dependency >

     < groupId >org.eclipse.paho</ groupId >

     < artifactId >org.eclipse.paho.client.mqttv3</ artifactId >

     < version >1.2.4</ version >

</ dependency >

spring: 

   mqtt:

     url: tcp://127.0.0.1:1883

     client-id: niubility-tiger

     username:

     password:

     topic: [/unify/test]

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;

}

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);

     }

}

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);

     }

}

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 ;

         }

     }

}