XA/mqtt-car
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: XA:dev_io_zc
Branches Tags
XA:master XA:dev_zc XA:td_h5 XA:dev_io_zc XA:dev XA:dev_lz
..
compare: XA:22480609050a74a444bdaefbd35f2dd9ee2ff5a9
Branches Tags
XA:dev_zc XA:td_h5 XA:dev_io_zc XA:dev XA:dev_lz XA:master

1 Commits
dev_io_zc ... 2248060905

Author SHA1 Message Date
zc
2248060905 h5提交 2026-05-12 10:49:59 +08:00
13 changed files with 239 additions and 388 deletions
Expand all files Collapse all files

35
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/config/SchedulerConfig.java
View File

@@ -1,35 +0,0 @@
package org.dromara.mica.mqtt.server.config;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* 定时任务线程池配置
* 解决默认定时任务线程池(大小为1)被阻塞导致所有定时任务无法执行的问题
*/
@Slf4j
@Configuration
@EnableScheduling
public class SchedulerConfig {
@Bean
public TaskScheduler taskScheduler() {
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
// 增加线程池大小，避免单线程阻塞影响所有定时任务
scheduler.setPoolSize(5);
scheduler.setThreadNamePrefix("mqtt-scheduler-");
// 等待所有任务完成后再关闭
scheduler.setWaitForTasksToCompleteOnShutdown(true);
// 等待时间
scheduler.setAwaitTerminationSeconds(60);
// 设置异常处理器，确保异常被记录但不中断定时任务
scheduler.setErrorHandler(throwable -> {
log.error("定时任务执行异常", throwable);
});
return scheduler;
}
}

16
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/entity/Equipment.java → example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/entity/IoLockEquipment.java
View File

@@ -10,8 +10,8 @@ import java.io.Serial;
import java.io.Serializable; import java.io.Serializable;
@Data @Data
@TableName("sys_equipment") @TableName("sys_io_lock_equipment")
public class Equipment implements Serializable { public class IoLockEquipment implements Serializable {
@Serial @Serial
private static final long serialVersionUID = 1L; private static final long serialVersionUID = 1L;
@@ -21,9 +21,6 @@ public class Equipment implements Serializable {
@TableId(type = IdType.AUTO) @TableId(type = IdType.AUTO)
private Long id; private Long id;
/** 所属产品Id */
private Long productId;
/** 设备名称 */ /** 设备名称 */
private String name; private String name;
@@ -36,15 +33,6 @@ public class Equipment implements Serializable {
/** 设备密码 */ /** 设备密码 */
private String password; private String password;
/** 设备区域 */
private Long spaceId;
/** 设备位置 */
private Long pointId;
/** 对接状态(0未对接 1对接成功) */
private Long state;
/** 设备状态(0在线 1离线) */ /** 设备状态(0在线 1离线) */
private String flag; private String flag;

225
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/listener/CarMessageListener.java
View File

@@ -47,28 +47,21 @@ public class CarMessageListener {
@Autowired @Autowired
RedisService redisService; RedisService redisService;
@Autowired // @Autowired
ICarParkRecordService carParkRecordService; // ICarParkRecordService carParkRecordService;
//
// @Autowired
// ICarPassRecordService carPassRecordService;
//
// @Autowired
// ICarParkItemService carParkItemService;
//
// @Autowired
// ICarPassGatherService carPassGatherService;
@Autowired // private static String key = "1234567898765432";
ICarPassRecordService carPassRecordService;
@Autowired // private String jinjiangUrl = "http://127.0.0.1:6609/";
ICarParkItemService carParkItemService;
@Autowired
ICarPassGatherService carPassGatherService;
@Autowired
private ServerService service;
private static String key = "1234567898765432";
//xa、jl、td、xj
private String jinjiangUrl = "http://127.0.0.1:6609/";
//zr
// private String jinjiangUrl = "http://192.168.155.42:6609/";
/** /**
* 心跳 * 心跳
@@ -91,7 +84,7 @@ public class CarMessageListener {
* @param topic * @param topic
* @param message * @param message
*/ */
@MqttServerFunction("device/${sn}/message/down/white_list_operator/reply") /*@MqttServerFunction("device/${sn}/message/down/white_list_operator/reply")
@Transactional(rollbackFor = Exception.class) @Transactional(rollbackFor = Exception.class)
public void white_list_operator_reply(String topic, byte[] message) { public void white_list_operator_reply(String topic, byte[] message) {
log.info("接收到车牌下发消息 -> Topic: {}, message: {}", topic, new String(message)); log.info("接收到车牌下发消息 -> Topic: {}, message: {}", topic, new String(message));
@@ -111,7 +104,7 @@ public class CarMessageListener {
carParkRecord.setSync("1"); carParkRecord.setSync("1");
} }
carParkRecordService.updateByClientId(carParkRecord); carParkRecordService.updateByClientId(carParkRecord);
} }*/
/** /**
* 车牌入场出场识别监听 * 车牌入场出场识别监听
@@ -123,98 +116,98 @@ public class CarMessageListener {
public void ivs_result(String topic, Map<String, String> topicVars, byte[] message) throws Exception { public void ivs_result(String topic, Map<String, String> topicVars, byte[] message) throws Exception {
String sn = topicVars.get("sn"); String sn = topicVars.get("sn");
log.info("接收到车辆识别消息 -> Topic: {}", topic); log.info("接收到车辆识别消息 -> Topic: {}", topic);
//
//如果开启了锁杆，且车位满了 // //如果开启了锁杆，且车位满了
CarParkItem carParkItem = carParkItemService.selectBySn(sn); // CarParkItem carParkItem = carParkItemService.selectBySn(sn);
long count = carPassGatherService.count(new QueryWrapper<CarPassGather>().eq("park_id", carParkItem.getParkId())); // long count = carPassGatherService.count(new QueryWrapper<CarPassGather>().eq("park_id", carParkItem.getParkId()));
if (StrUtil.equals(carParkItem.getWay(), "0") //入场 // if (StrUtil.equals(carParkItem.getWay(), "0") //入场
&& carParkItem.getAutoLock() == 1 //开启了锁杆 // && carParkItem.getAutoLock() == 1 //开启了锁杆
&& count >= carParkItem.getCarsNum()) { //车位满了 // && count >= carParkItem.getCarsNum()) { //车位满了
log.info("车位满了，不记录入场数据"); // log.info("车位满了，不记录入场数据");
return; // return;
} // }
//
String data = new String(message, StandardCharsets.UTF_8); // String data = new String(message, StandardCharsets.UTF_8);
JSONObject jsonObject = JSONObject.parseObject(data); // JSONObject jsonObject = JSONObject.parseObject(data);
JSONObject payload = jsonObject.getJSONObject("payload"); // JSONObject payload = jsonObject.getJSONObject("payload");
String id = jsonObject.getString("id"); // String id = jsonObject.getString("id");
JSONObject alarmInfoPlate = payload.getJSONObject("AlarmInfoPlate"); // JSONObject alarmInfoPlate = payload.getJSONObject("AlarmInfoPlate");
JSONObject result = alarmInfoPlate.getJSONObject("result"); // JSONObject result = alarmInfoPlate.getJSONObject("result");
JSONObject plateResult = result.getJSONObject("PlateResult"); // JSONObject plateResult = result.getJSONObject("PlateResult");
JSONObject carBrand = plateResult.getJSONObject("car_brand"); // JSONObject carBrand = plateResult.getJSONObject("car_brand");
String license = plateResult.getString("license"); // String license = plateResult.getString("license");
String colorType = plateResult.getString("colorType"); // String colorType = plateResult.getString("colorType");
String str = ""; // String str = "";
if ("5".equals(colorType)) { // if ("5".equals(colorType)) {
str = AESUtil.decrptyAES_ECB(license, key).substring(0, 20); // str = AESUtil.decrptyAES_ECB(license, key).substring(0, 20);
} else { // } else {
str = AESUtil.decrptyAES_ECB(license, key).substring(0, 18); // str = AESUtil.decrptyAES_ECB(license, key).substring(0, 18);
} // }
//
license = AESUtil.UTF8decode(str); // license = AESUtil.UTF8decode(str);
log.info("解密前车牌：{}，解谜后的车牌:{}", plateResult.getString("license"), license); // log.info("解密前车牌：{}，解谜后的车牌:{}", plateResult.getString("license"), license);
//
//保存通行记录 // //保存通行记录
CarPassRecord record = new CarPassRecord(); // CarPassRecord record = new CarPassRecord();
record.setCarColor(plateResult.getString("carColor")); // record.setCarColor(plateResult.getString("carColor"));
record.setColorType(colorType); // record.setColorType(colorType);
record.setDirection(plateResult.getString("direction")); // record.setDirection(plateResult.getString("direction"));
record.setLicense(license); // record.setLicense(license);
record.setDataType("0"); // record.setDataType("0");
Map<String, Object> paramMap = new HashMap<>(); // Map<String, Object> paramMap = new HashMap<>();
paramMap.put("url", plateResult.getString("full_image_content")); // paramMap.put("url", plateResult.getString("full_image_content"));
String repose = HttpUtil.createPost(jinjiangUrl + "file/uploadMinioCarBase64") // String repose = HttpUtil.createPost(jinjiangUrl + "file/uploadMinioCarBase64")
.body(JSON.toJSONString(paramMap)) // .body(JSON.toJSONString(paramMap))
.execute() // .execute()
.body(); // .body();
log.info("imgRsp:{}", repose); // log.info("imgRsp:{}", repose);
JSONObject imgRsp = JSONObject.parseObject(repose); // JSONObject imgRsp = JSONObject.parseObject(repose);
if (null != imgRsp && imgRsp.getInteger("code") == 200) { // if (null != imgRsp && imgRsp.getInteger("code") == 200) {
record.setUrl(imgRsp.getJSONObject("data").getString("url")); // record.setUrl(imgRsp.getJSONObject("data").getString("url"));
} // }
if (plateResult.containsKey("start_time")) { // if (plateResult.containsKey("start_time")) {
log.info("拿到了时间：{}", plateResult.getLong("start_time")); // log.info("拿到了时间：{}", plateResult.getLong("start_time"));
record.setPassTime(DateUtil.date(plateResult.getLong("start_time"))); // record.setPassTime(DateUtil.date(plateResult.getLong("start_time")));
} else { // } else {
log.info("没有拿到时间，默认当前时间"); // log.info("没有拿到时间，默认当前时间");
record.setPassTime(DateUtil.date(new Date())); // record.setPassTime(DateUtil.date(new Date()));
} // }
record.setSn(sn); // record.setSn(sn);
record.setUniqueNo(id); // record.setUniqueNo(id);
record.setParkId(carParkItem.getParkId()); // record.setParkId(carParkItem.getParkId());
record.setTriggerType(plateResult.getString("triggerType")); // record.setTriggerType(plateResult.getString("triggerType"));
record.setType(carBrand.getString("type")); // record.setType(carBrand.getString("type"));
carPassRecordService.save(record); // carPassRecordService.save(record);
//
//保存/删除在场数据 // //保存/删除在场数据
boolean isExist = carPassGatherService.exists(new QueryWrapper<CarPassGather>().eq("license", license).eq("park_id", carParkItem.getParkId())); // boolean isExist = carPassGatherService.exists(new QueryWrapper<CarPassGather>().eq("license", license).eq("park_id", carParkItem.getParkId()));
//入场新增数据 // //入场新增数据
if (StrUtil.equals(carParkItem.getWay(), "0") && !isExist) { // if (StrUtil.equals(carParkItem.getWay(), "0") && !isExist) {
CarPassGather carPassGather = new CarPassGather(); // CarPassGather carPassGather = new CarPassGather();
carPassGather.setLicense(license); // carPassGather.setLicense(license);
carPassGather.setParkId(carParkItem.getParkId()); // carPassGather.setParkId(carParkItem.getParkId());
carPassGather.setArea(carParkItem.getArea()); // carPassGather.setArea(carParkItem.getArea());
carPassGather.setJoinTime(record.getPassTime()); // carPassGather.setJoinTime(record.getPassTime());
carPassGather.setSn(sn); // carPassGather.setSn(sn);
carPassGatherService.save(carPassGather); // carPassGatherService.save(carPassGather);
//
//最后一辆车进场 // //最后一辆车进场
if (carParkItem.getAutoLock() == 1 && (count + 1) >= carParkItem.getCarsNum()) { // if (carParkItem.getAutoLock() == 1 && (count + 1) >= carParkItem.getCarsNum()) {
List<String> snList = carParkItemService.selectSnByParkId(carParkItem.getParkId()); // List<String> snList = carParkItemService.selectSnByParkId(carParkItem.getParkId());
for (String s : snList) { // for (String s : snList) {
service.locked(s, 2, 0); // service.locked(s, 2, 0);
} // }
} // }
} else if (StrUtil.equals(carParkItem.getWay(), "1") && isExist) {//出场删除数据 // } else if (StrUtil.equals(carParkItem.getWay(), "1") && isExist) {//出场删除数据
carPassGatherService.deleteByLicense(license); // carPassGatherService.deleteByLicense(license);
//出去一辆车，有空位 // //出去一辆车，有空位
if (carParkItem.getAutoLock() == 1 && count == carParkItem.getCarsNum()) { // if (carParkItem.getAutoLock() == 1 && count == carParkItem.getCarsNum()) {
List<String> snList = carParkItemService.selectSnByParkId(carParkItem.getParkId()); // List<String> snList = carParkItemService.selectSnByParkId(carParkItem.getParkId());
for (String s : snList) { // for (String s : snList) {
service.locked(s, 0, 0); // service.locked(s, 0, 0);
} // }
} // }
} // }
} }
/** /**
@@ -223,7 +216,7 @@ public class CarMessageListener {
* @param topic * @param topic
* @param message * @param message
*/ */
@MqttServerFunction("device/${sn}/message/up/ivs_result_offline") /*@MqttServerFunction("device/${sn}/message/up/ivs_result_offline")
public void ivs_result_offline(String topic, Map<String, String> topicVars, byte[] message) throws Exception { public void ivs_result_offline(String topic, Map<String, String> topicVars, byte[] message) throws Exception {
String sn = topicVars.get("sn"); String sn = topicVars.get("sn");
log.info("接收到车辆离线识别消息 -> Topic: {}", topic); log.info("接收到车辆离线识别消息 -> Topic: {}", topic);
@@ -278,7 +271,7 @@ public class CarMessageListener {
record.setTriggerType(plateResult.getString("triggerType")); record.setTriggerType(plateResult.getString("triggerType"));
record.setType(carBrand.getString("type")); record.setType(carBrand.getString("type"));
carPassRecordService.save(record); carPassRecordService.save(record);
} }*/
/** /**
* IO输出事件监听 * IO输出事件监听

5
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/listener/MqttConnectStatusListener2.java
View File

@@ -16,13 +16,10 @@
package org.dromara.mica.mqtt.server.listener; package org.dromara.mica.mqtt.server.listener;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.core.server.event.IMqttConnectStatusListener; import org.dromara.mica.mqtt.core.server.event.IMqttConnectStatusListener;
import org.dromara.mica.mqtt.server.constant.CacheConstants; import org.dromara.mica.mqtt.server.constant.CacheConstants;
import org.dromara.mica.mqtt.server.entity.Equipment;
import org.dromara.mica.mqtt.server.enums.FlagEnums; import org.dromara.mica.mqtt.server.enums.FlagEnums;
import org.dromara.mica.mqtt.server.mapper.EquipmentMapper;
import org.dromara.mica.mqtt.server.redis.RedisService; import org.dromara.mica.mqtt.server.redis.RedisService;
import org.dromara.mica.mqtt.server.service.IEquipmentService; import org.dromara.mica.mqtt.server.service.IEquipmentService;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
@@ -30,8 +27,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
import org.tio.core.ChannelContext; import org.tio.core.ChannelContext;
import java.util.concurrent.TimeUnit;
/** /**
* mqtt 连接状态 * mqtt 连接状态
* *

5
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/mapper/EquipmentMapper.java
View File

@@ -2,10 +2,9 @@ package org.dromara.mica.mqtt.server.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Mapper;
import org.dromara.mica.mqtt.server.entity.CarInfo; import org.dromara.mica.mqtt.server.entity.IoLockEquipment;
import org.dromara.mica.mqtt.server.entity.Equipment;
@Mapper @Mapper
public interface EquipmentMapper extends BaseMapper<Equipment> { public interface EquipmentMapper extends BaseMapper<IoLockEquipment> {
} }

22
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/redis/RedisService.java
View File

@@ -80,11 +80,7 @@ public class RedisService {
*/ */
public <T> void setCacheObject(final String key, final T value, final Long timeout, final TimeUnit timeUnit) public <T> void setCacheObject(final String key, final T value, final Long timeout, final TimeUnit timeUnit)
{ {
try { redisTemplate.opsForValue().set(key, value, timeout, timeUnit);
redisTemplate.opsForValue().set(key, value, timeout, timeUnit);
} catch (Exception e) {
log.error("Redis setCacheObject error: key={}", key, e);
}
} }
/** /**
@@ -131,12 +127,7 @@ public class RedisService {
*/ */
public Boolean hasKey(String key) public Boolean hasKey(String key)
{ {
try { return redisTemplate.hasKey(key);
return redisTemplate.hasKey(key);
} catch (Exception e) {
log.error("Redis hasKey error: key={}", key, e);
return false;
}
} }
/** /**
@@ -147,13 +138,8 @@ public class RedisService {
*/ */
public <T> T getCacheObject(final String key) public <T> T getCacheObject(final String key)
{ {
try { ValueOperations<String, T> operation = redisTemplate.opsForValue();
ValueOperations<String, T> operation = redisTemplate.opsForValue(); return operation.get(key);
return operation.get(key);
} catch (Exception e) {
log.error("Redis getCacheObject error: key={}", key, e);
return null;
}
} }
/** /**

8
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/service/IEquipmentService.java
View File

@@ -1,15 +1,15 @@
package org.dromara.mica.mqtt.server.service; package org.dromara.mica.mqtt.server.service;
import com.baomidou.mybatisplus.extension.service.IService; import com.baomidou.mybatisplus.extension.service.IService;
import org.dromara.mica.mqtt.server.entity.Equipment; import org.dromara.mica.mqtt.server.entity.IoLockEquipment;
import java.util.List; import java.util.List;
public interface IEquipmentService extends IService<Equipment> { public interface IEquipmentService extends IService<IoLockEquipment> {
Equipment selectEquipmentBySn(String sn); IoLockEquipment selectEquipmentBySn(String sn);
List<Equipment> selectAllSnFlag(); List<IoLockEquipment> selectAllSnFlag();
void updateFlag(String sn, String flag); void updateFlag(String sn, String flag);
} }

16
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/service/impl/EquipmentServiceImpl.java
View File

@@ -2,7 +2,7 @@ package org.dromara.mica.mqtt.server.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.dromara.mica.mqtt.server.entity.Equipment; import org.dromara.mica.mqtt.server.entity.IoLockEquipment;
import org.dromara.mica.mqtt.server.mapper.EquipmentMapper; import org.dromara.mica.mqtt.server.mapper.EquipmentMapper;
import org.dromara.mica.mqtt.server.service.IEquipmentService; import org.dromara.mica.mqtt.server.service.IEquipmentService;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
@@ -11,25 +11,25 @@ import org.springframework.stereotype.Service;
import java.util.List; import java.util.List;
@Service @Service
public class EquipmentServiceImpl extends ServiceImpl<EquipmentMapper, Equipment> implements IEquipmentService { public class EquipmentServiceImpl extends ServiceImpl<EquipmentMapper, IoLockEquipment> implements IEquipmentService {
@Autowired @Autowired
EquipmentMapper equipmentMapper; EquipmentMapper equipmentMapper;
@Override @Override
public Equipment selectEquipmentBySn(String sn) { public IoLockEquipment selectEquipmentBySn(String sn) {
return equipmentMapper.selectOne(new QueryWrapper<Equipment>().eq("sequence", sn).eq("product_id", 4L).last("limit 1")); return equipmentMapper.selectOne(new QueryWrapper<IoLockEquipment>().eq("sequence", sn).last("limit 1"));
} }
@Override @Override
public List<Equipment> selectAllSnFlag() { public List<IoLockEquipment> selectAllSnFlag() {
return equipmentMapper.selectList(new QueryWrapper<Equipment>().eq("product_id", 4L).select("sequence", "flag")); return equipmentMapper.selectList(new QueryWrapper<IoLockEquipment>().select("sequence", "flag"));
} }
@Override @Override
public void updateFlag(String sn, String flag) { public void updateFlag(String sn, String flag) {
Equipment equipment = new Equipment(); IoLockEquipment equipment = new IoLockEquipment();
equipment.setFlag(flag); equipment.setFlag(flag);
equipmentMapper.update(equipment, new QueryWrapper<Equipment>().eq("sequence", sn).eq("product_id", 4L)); equipmentMapper.update(equipment, new QueryWrapper<IoLockEquipment>().eq("sequence", sn));
} }
} }

46
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/task/HeartbeatOnLineTask.java
View File

@@ -1,11 +1,9 @@
package org.dromara.mica.mqtt.server.task; package org.dromara.mica.mqtt.server.task;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.dromara.mica.mqtt.server.constant.CacheConstants; import org.dromara.mica.mqtt.server.constant.CacheConstants;
import org.dromara.mica.mqtt.server.entity.Equipment; import org.dromara.mica.mqtt.server.entity.IoLockEquipment;
import org.dromara.mica.mqtt.server.enums.FlagEnums; import org.dromara.mica.mqtt.server.enums.FlagEnums;
import org.dromara.mica.mqtt.server.mapper.EquipmentMapper;
import org.dromara.mica.mqtt.server.redis.RedisService; import org.dromara.mica.mqtt.server.redis.RedisService;
import org.dromara.mica.mqtt.server.service.IEquipmentService; import org.dromara.mica.mqtt.server.service.IEquipmentService;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
@@ -30,32 +28,28 @@ public class HeartbeatOnLineTask {
@Scheduled(fixedRate = 15 * 1000) @Scheduled(fixedRate = 15 * 1000)
public void run() { public void run() {
// 添加异常捕获，防止单个任务异常导致定时任务线程阻塞 log.info("===========心跳检测=============");
try { //查询车辆摄像头的编码和在线状态
log.info("===========心跳检测============="); List<IoLockEquipment> equipment = equipmentService.selectAllSnFlag();
//查询车辆摄像头的编码和在线状态
List<Equipment> equipment = equipmentService.selectAllSnFlag();
for (Equipment equip : equipment) { for (IoLockEquipment equip : equipment) {
//缓存中有该设备心跳key //缓存中有该设备心跳key
if (redisService.hasKey(CacheConstants.EQUIPMENT_HEARTBEAT + equip.getSequence())) { if (redisService.hasKey(CacheConstants.EQUIPMENT_HEARTBEAT + equip.getSequence())) {
String flag = redisService.getCacheObject(CacheConstants.EQUIPMENT_HEARTBEAT + equip.getSequence()); String flag = redisService.getCacheObject(CacheConstants.EQUIPMENT_HEARTBEAT + equip.getSequence());
//有心跳，但是和数据库设备在线状态不一致 //有心跳，但是和数据库设备在线状态不一致
if (!StrUtil.equals(flag, equip.getFlag())) { if (!StrUtil.equals(flag, equip.getFlag())) {
log.info("设备：{}，flag:{}", equip.getSequence(), flag); log.info("设备：{}，flag:{}", equip.getSequence(), flag);
equipmentService.updateFlag(equip.getSequence(), flag); equipmentService.updateFlag(equip.getSequence(), flag);
} }
} else { } else {
//没有心跳上传，且设备在线,将设备置为离线 //没有心跳上传，且设备在线,将设备置为离线
if (StrUtil.equals(FlagEnums.ONLINE.getCode(), equip.getFlag())) { if (StrUtil.equals(FlagEnums.ONLINE.getCode(), equip.getFlag())) {
log.info("设备：{}，无心跳离线", equip.getSequence()); log.info("设备：{}，无心跳离线", equip.getSequence());
equipmentService.updateFlag(equip.getSequence(), FlagEnums.OFFLINE.getCode()); equipmentService.updateFlag(equip.getSequence(), FlagEnums.OFFLINE.getCode());
}
} }
} }
} catch (Exception e) {
// 捕获所有异常并记录日志，确保定时任务能继续执行
log.error("HeartbeatOnLineTask 定时任务执行异常", e);
} }
} }
} }

173
example/mqtt-car/src/main/java/org/dromara/mica/mqtt/server/task/PlatePublishTask.java
View File

@@ -48,100 +48,95 @@ public class PlatePublishTask {
/** /**
* 定时查询数据库需要下发的车辆进行下发 * 定时查询数据库需要下发的车辆进行下发
*/ */
@Scheduled(fixedDelay = 8 * 1000) // @Scheduled(fixedDelay = 8 * 1000)
public void run() { public void run() {
// 添加异常捕获，防止单个任务异常导致定时任务线程阻塞 String whiteUrl = "device/%s/message/down/white_list_operator";
try {
String whiteUrl = "device/%s/message/down/white_list_operator";
//新增&编辑车牌 //新增&编辑车牌
CarInfo carInfo = new CarInfo(); CarInfo carInfo = new CarInfo();
carInfo.setDelFlag("0"); carInfo.setDelFlag("0");
carInfo.setSync("0"); carInfo.setSync("0");
List<CarInfo> carInfoList = carInfoService.selectCarInfoList(carInfo); List<CarInfo> carInfoList = carInfoService.selectCarInfoList(carInfo);
if (CollUtil.isNotEmpty(carInfoList)) { if (CollUtil.isNotEmpty(carInfoList)) {
//向设备下发车辆信息 //向设备下发车辆信息
for (CarInfo car : carInfoList) { for (CarInfo car : carInfoList) {
//如果该设备没有心跳，不下发车牌信息 //如果该设备没有心跳，不下发车牌信息
if (!redisService.hasKey(CacheConstants.EQUIPMENT_HEARTBEAT + car.getSn())) { if (!redisService.hasKey(CacheConstants.EQUIPMENT_HEARTBEAT + car.getSn())) {
continue; continue;
}
String topic = String.format(whiteUrl, car.getSn());
WhiteListOperatorPO po = new WhiteListOperatorPO();
String uuid = "add_" + UuidUtil.getUuid();
po.setId(uuid);
po.setSn(car.getSn());
po.setName("white_list_operator");
//构造数据
po.setPayload(buildPayload(car));
//更新car_park_record表的clientId，用于回执消息更新下发状态
//必须在发送前更新，不然会导致发送后，还没更新完数据库，回执已经收到，无法更新数据，导致持续下发车牌
CarParkRecord carParkRecord = new CarParkRecord();
carParkRecord.setId(car.getCarParkRecordId());
carParkRecord.setClientId(uuid);
carParkRecordService.updateById(carParkRecord);
//发布车牌到设备（协议只能单条发布）
boolean publish = mqttServer.publish(car.getSn(), topic, JSON.toJSONString(po).getBytes(StandardCharsets.UTF_8));
log.info("定时任务下发车牌topic:{},报文数据：{},发送结果：{}", topic, JSON.toJSONString(po), publish);
} }
String topic = String.format(whiteUrl, car.getSn());
WhiteListOperatorPO po = new WhiteListOperatorPO();
String uuid = "add_" + UuidUtil.getUuid();
po.setId(uuid);
po.setSn(car.getSn());
po.setName("white_list_operator");
//构造数据
po.setPayload(buildPayload(car));
//更新car_park_record表的clientId，用于回执消息更新下发状态
//必须在发送前更新，不然会导致发送后，还没更新完数据库，回执已经收到，无法更新数据，导致持续下发车牌
CarParkRecord carParkRecord = new CarParkRecord();
carParkRecord.setId(car.getCarParkRecordId());
carParkRecord.setClientId(uuid);
carParkRecordService.updateById(carParkRecord);
//发布车牌到设备（协议只能单条发布）
boolean publish = mqttServer.publish(car.getSn(), topic, JSON.toJSONString(po).getBytes(StandardCharsets.UTF_8));
log.info("定时任务下发车牌topic:{},报文数据：{},发送结果：{}", topic, JSON.toJSONString(po), publish);
} }
//删除车牌
CarInfo carDel = new CarInfo();
carDel.setDelFlag("2");
carDel.setSync("1");
List<CarInfo> carInfos = carInfoService.selectCarInfoList(carDel);
if (CollUtil.isNotEmpty(carInfos)) {
//需要下发的设备
List<String> snList = carInfos.stream().map(CarInfo::getSn).distinct().toList();
for (String sn : snList) {
if (!redisService.hasKey(CacheConstants.EQUIPMENT_HEARTBEAT + sn)) {
log.error("删除车牌，设备：{}，无心跳", sn);
continue;
}
//车辆授权列表
List<CarParkRecord> carParkRecords = new ArrayList<>();
String topic = String.format(whiteUrl, sn);
WhiteListOperatorPO po = new WhiteListOperatorPO();
String uuid = "del_" + UuidUtil.getUuid();
po.setId(uuid);
po.setSn(sn);
po.setName("white_list_operator");
//构造内层数据
List<String> plates = new ArrayList<>();
for (CarInfo car : carInfos) {
plates.add(car.getPlate());
if (StrUtil.equals(car.getSn(), sn)) {
CarParkRecord carParkRecord = new CarParkRecord();
carParkRecord.setClientId(uuid);
carParkRecord.setId(car.getCarParkRecordId());
carParkRecords.add(carParkRecord);
}
}
po.setPayload(buildPayloadDel(plates));
//更新车辆授权信息
if (CollUtil.isNotEmpty(carParkRecords)) {
carParkRecordService.saveOrUpdateBatch(carParkRecords);
}
//发布车牌到设备
boolean publish = mqttServer.publish(sn, topic, JSON.toJSONString(po).getBytes(StandardCharsets.UTF_8));
log.info("定时任务删除车牌topic:{},报文数据：{},发送结果：{}", topic, JSON.toJSONString(po), publish);
}
}
} catch (Exception e) {
// 捕获所有异常并记录日志，确保定时任务能继续执行
log.error("PlatePublishTask 定时任务执行异常", e);
} }
//删除车牌
CarInfo carDel = new CarInfo();
carDel.setDelFlag("2");
carDel.setSync("1");
List<CarInfo> carInfos = carInfoService.selectCarInfoList(carDel);
if (CollUtil.isNotEmpty(carInfos)) {
//需要下发的设备
List<String> snList = carInfos.stream().map(CarInfo::getSn).distinct().toList();
for (String sn : snList) {
if (!redisService.hasKey(CacheConstants.EQUIPMENT_HEARTBEAT + sn)) {
log.error("删除车牌，设备：{}，无心跳", sn);
continue;
}
//车辆授权列表
List<CarParkRecord> carParkRecords = new ArrayList<>();
String topic = String.format(whiteUrl, sn);
WhiteListOperatorPO po = new WhiteListOperatorPO();
String uuid = "del_" + UuidUtil.getUuid();
po.setId(uuid);
po.setSn(sn);
po.setName("white_list_operator");
//构造内层数据
List<String> plates = new ArrayList<>();
for (CarInfo car : carInfos) {
plates.add(car.getPlate());
if (StrUtil.equals(car.getSn(), sn)) {
CarParkRecord carParkRecord = new CarParkRecord();
carParkRecord.setClientId(uuid);
carParkRecord.setId(car.getCarParkRecordId());
carParkRecords.add(carParkRecord);
}
}
po.setPayload(buildPayloadDel(plates));
//更新车辆授权信息
if (CollUtil.isNotEmpty(carParkRecords)) {
carParkRecordService.saveOrUpdateBatch(carParkRecords);
}
//发布车牌到设备
boolean publish = mqttServer.publish(sn, topic, JSON.toJSONString(po).getBytes(StandardCharsets.UTF_8));
log.info("定时任务删除车牌topic:{},报文数据：{},发送结果：{}", topic, JSON.toJSONString(po), publish);
}
}
} }

23
example/mqtt-car/src/main/resources/application-dev.yml
View File

@@ -1,33 +1,12 @@
spring: spring:
datasource: datasource:
driver-class-name: com.mysql.cj.jdbc.Driver driver-class-name: com.mysql.cj.jdbc.Driver
url: jdbc:mysql://127.0.0.1:3306/td_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&connectTimeout=5000&socketTimeout=30000 url: jdbc:mysql://127.0.0.1:3306/td_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
username: root username: root
password: root password: root
# Druid 连接池配置
druid:
max-active: 20
max-wait: 60000
min-idle: 5
initial-size: 5
validation-query: SELECT 1
test-while-idle: true
test-on-borrow: false
test-on-return: false
time-between-eviction-runs-millis: 60000
min-evictable-idle-time-millis: 300000
data: data:
redis: redis:
# host: 127.0.0.1
host: 192.168.2.30 host: 192.168.2.30
password: redis2025 password: redis2025
database: 5 database: 5
port: 6379 port: 6379
# Redis 超时配置，防止连接阻塞
timeout: 5000ms
lettuce:
pool:
max-active: 8
max-idle: 8
min-idle: 2
max-wait: 5000ms

48
example/mqtt-car/src/main/resources/application-prod.yml
View File

@@ -1,52 +1,14 @@
spring: spring:
datasource: datasource:
driver-class-name: com.mysql.cj.jdbc.Driver driver-class-name: com.mysql.cj.jdbc.Driver
#xa
# url: jdbc:mysql://127.0.0.1:3306/xa_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&connectTimeout=5000&socketTimeout=30000
# username: root
# password: Xahg2024.
#jl
# url: jdbc:mysql://127.0.0.1:3306/jl_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&connectTimeout=5000&socketTimeout=30000
# username: root
# password: JL202509jj
#td #td
# url: jdbc:mysql://127.0.0.1:3306/td_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&connectTimeout=5000&socketTimeout=30000 url: jdbc:mysql://192.168.251.16:3306/td_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
# username: root
# password: td@JJ2024
#zr
# url: jdbc:mysql://192.168.155.42:3306/zr_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&connectTimeout=5000&socketTimeout=30000
# username: root
# password: zr202407.J
#xj
url: jdbc:mysql://127.0.0.1:3306/xj_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&connectTimeout=5000&socketTimeout=30000
username: root username: root
password: XjJN2024! password: td@JJ2024
# Druid 连接池配置
druid:
max-active: 20
max-wait: 60000
min-idle: 5
initial-size: 5
validation-query: SELECT 1
test-while-idle: true
test-on-borrow: false
test-on-return: false
time-between-eviction-runs-millis: 60000
min-evictable-idle-time-millis: 300000
data: data:
redis: redis:
#zr host: 192.168.251.16
# host: 192.168.155.42
# xa、jl、td、xj
host: 127.0.0.1
port: 6379 port: 6379
password: password:
database: 1 database: 2
# Redis 超时配置，防止连接阻塞
timeout: 5000ms
lettuce:
pool:
max-active: 8
max-idle: 8
min-idle: 2
max-wait: 5000ms

5
example/mqtt-car/src/main/resources/application.yml
View File

@@ -127,8 +127,3 @@ logging:
server: info # t-io 服务端默认日志 server: info # t-io 服务端默认日志
org.tio: info # t-io 服务端默认日志 org.tio: info # t-io 服务端默认日志
org.dromara.mica.mqtt: info # mica-mqtt 日志 org.dromara.mica.mqtt: info # mica-mqtt 日志
org.dromara.mica.mqtt.server.task: debug # 定时任务日志级别
# 定时任务线程池配置
scheduler:
pool-size: 5