Compare commits
6 Commits
2248060905
...
td_h5
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d43acc31d9 | ||
|
|
58b76371c5 | ||
|
|
e9a698c728 | ||
|
|
118729ccb2 | ||
|
|
4c41f6f77e | ||
|
|
90003d79c5 |
@@ -0,0 +1,35 @@
|
|||||||
|
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;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -50,10 +50,12 @@ public class ServerController {
|
|||||||
}
|
}
|
||||||
|
|
||||||
//手动触发抬杆
|
//手动触发抬杆
|
||||||
@PostMapping("/openFloodgate")
|
@PostMapping("/gpio_out")
|
||||||
public boolean openFloodgate(@RequestBody JSONObject js) {
|
public boolean gpio_out(@RequestBody JSONObject js) {
|
||||||
String sn = js.getString("sn");
|
String sn = js.getString("sn");
|
||||||
return service.openFloodgate(sn);
|
Integer io = js.getInteger("io");
|
||||||
|
Integer value = js.getInteger("value");
|
||||||
|
return service.gpio_out(sn, io, value);
|
||||||
}
|
}
|
||||||
|
|
||||||
@PostMapping("/set_io_lock_status")
|
@PostMapping("/set_io_lock_status")
|
||||||
@@ -61,7 +63,7 @@ public class ServerController {
|
|||||||
String sn = js.getString("sn");
|
String sn = js.getString("sn");
|
||||||
Integer status = js.getInteger("status");
|
Integer status = js.getInteger("status");
|
||||||
Integer ioout = js.getInteger("ioout");
|
Integer ioout = js.getInteger("ioout");
|
||||||
boolean publish = service.locked(sn, status,ioout);
|
boolean publish = service.set_io_lock_status(sn, status,ioout);
|
||||||
JSONObject jsonObject = new JSONObject();
|
JSONObject jsonObject = new JSONObject();
|
||||||
if (publish) {
|
if (publish) {
|
||||||
jsonObject.put("code", 200);
|
jsonObject.put("code", 200);
|
||||||
@@ -71,7 +73,7 @@ public class ServerController {
|
|||||||
return jsonObject;
|
return jsonObject;
|
||||||
}
|
}
|
||||||
|
|
||||||
//常开
|
//开闸
|
||||||
@PostMapping("/set_io_lock_open")
|
@PostMapping("/set_io_lock_open")
|
||||||
public JSONObject lockOpen(@RequestBody JSONObject js) {
|
public JSONObject lockOpen(@RequestBody JSONObject js) {
|
||||||
String sn = js.getString("sn");
|
String sn = js.getString("sn");
|
||||||
@@ -85,7 +87,7 @@ public class ServerController {
|
|||||||
return jsonObject;
|
return jsonObject;
|
||||||
}
|
}
|
||||||
|
|
||||||
//常关
|
//关闸
|
||||||
@PostMapping("/set_io_lock_close")
|
@PostMapping("/set_io_lock_close")
|
||||||
public JSONObject lockClose(@RequestBody JSONObject js) {
|
public JSONObject lockClose(@RequestBody JSONObject js) {
|
||||||
String sn = js.getString("sn");
|
String sn = js.getString("sn");
|
||||||
|
|||||||
@@ -74,7 +74,7 @@ public class CarMessageListener {
|
|||||||
String sn = topicVars.get("sn");
|
String sn = topicVars.get("sn");
|
||||||
log.info("接收到来自客户端 [{}] 的心跳消息 -> Topic: {}", sn, topic);
|
log.info("接收到来自客户端 [{}] 的心跳消息 -> Topic: {}", sn, topic);
|
||||||
|
|
||||||
// 更新客户端的最后心跳
|
// 更新客户端的最后心跳
|
||||||
redisService.setCacheObject(CacheConstants.EQUIPMENT_HEARTBEAT + sn, FlagEnums.ONLINE.getCode(), CacheConstants.OFFLINE_THRESHOLD, TimeUnit.MILLISECONDS);
|
redisService.setCacheObject(CacheConstants.EQUIPMENT_HEARTBEAT + sn, FlagEnums.ONLINE.getCode(), CacheConstants.OFFLINE_THRESHOLD, TimeUnit.MILLISECONDS);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -313,6 +313,19 @@ public class CarMessageListener {
|
|||||||
JSONObject jsonObject = JSONObject.parseObject(data);
|
JSONObject jsonObject = JSONObject.parseObject(data);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 发布获取IO状态监听
|
||||||
|
*
|
||||||
|
* @param topic
|
||||||
|
* @param message
|
||||||
|
*/
|
||||||
|
@MqttServerFunction("device/${sn}/message/down/get_io_status/reply")
|
||||||
|
public void get_io_status(String topic, byte[] message) {
|
||||||
|
log.info("发布获取IO状态监听消息 -> Topic: {}, message: {}", topic, new String(message));
|
||||||
|
String data = new String(message, StandardCharsets.UTF_8);
|
||||||
|
JSONObject jsonObject = JSONObject.parseObject(data);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 订阅离线数据数量
|
* 订阅离线数据数量
|
||||||
|
|||||||
@@ -80,7 +80,11 @@ 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)
|
||||||
{
|
{
|
||||||
redisTemplate.opsForValue().set(key, value, timeout, timeUnit);
|
try {
|
||||||
|
redisTemplate.opsForValue().set(key, value, timeout, timeUnit);
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error("Redis setCacheObject error: key={}", key, e);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -127,7 +131,12 @@ public class RedisService {
|
|||||||
*/
|
*/
|
||||||
public Boolean hasKey(String key)
|
public Boolean hasKey(String key)
|
||||||
{
|
{
|
||||||
return redisTemplate.hasKey(key);
|
try {
|
||||||
|
return redisTemplate.hasKey(key);
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error("Redis hasKey error: key={}", key, e);
|
||||||
|
return false;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -138,8 +147,13 @@ public class RedisService {
|
|||||||
*/
|
*/
|
||||||
public <T> T getCacheObject(final String key)
|
public <T> T getCacheObject(final String key)
|
||||||
{
|
{
|
||||||
ValueOperations<String, T> operation = redisTemplate.opsForValue();
|
try {
|
||||||
return operation.get(key);
|
ValueOperations<String, T> operation = redisTemplate.opsForValue();
|
||||||
|
return operation.get(key);
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error("Redis getCacheObject error: key={}", key, e);
|
||||||
|
return null;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -62,8 +62,7 @@ public class ServerService {
|
|||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public boolean gpio_out(String sn, Integer io, Integer value) {
|
||||||
public boolean openFloodgate(String sn) {
|
|
||||||
String operUrl = "device/%s/message/down/gpio_out";
|
String operUrl = "device/%s/message/down/gpio_out";
|
||||||
String topic = String.format(operUrl, sn);
|
String topic = String.format(operUrl, sn);
|
||||||
OpenFloodgatePO openFloodgatePO = new OpenFloodgatePO();
|
OpenFloodgatePO openFloodgatePO = new OpenFloodgatePO();
|
||||||
@@ -71,16 +70,26 @@ public class ServerService {
|
|||||||
openFloodgatePO.setId(uuid);
|
openFloodgatePO.setId(uuid);
|
||||||
openFloodgatePO.setSn(sn);
|
openFloodgatePO.setSn(sn);
|
||||||
openFloodgatePO.setName("gpio_out");
|
openFloodgatePO.setName("gpio_out");
|
||||||
LocalDateTime now = LocalDateTime.now();
|
openFloodgatePO.setTimestamp(System.currentTimeMillis() / 1000);
|
||||||
long timestamp = now.atZone(ZoneId.systemDefault()).toEpochSecond();
|
openFloodgatePO.setPayload(this.buildGpioOut(io, value));
|
||||||
openFloodgatePO.setTimestamp(timestamp);
|
|
||||||
openFloodgatePO.setPayload(this.buildPayloadOpen());
|
|
||||||
boolean result = server.publish(sn,topic, JSON.toJSONString(openFloodgatePO).getBytes(StandardCharsets.UTF_8));
|
boolean result = server.publish(sn,topic, JSON.toJSONString(openFloodgatePO).getBytes(StandardCharsets.UTF_8));
|
||||||
log.info("抬杠设备编码:{},result:{},请求体{}", sn,result,openFloodgatePO);
|
log.info("gpio_out设备编码:{},result:{},请求体{}", sn,result,openFloodgatePO);
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean locked(String sn, Integer status, Integer ioout) {
|
|
||||||
|
private JSONObject buildGpioOut(Integer io, Integer value) {
|
||||||
|
JSONObject body = new JSONObject();
|
||||||
|
body.put("delay", 500);
|
||||||
|
body.put("io", io);
|
||||||
|
body.put("value", value);
|
||||||
|
JSONObject payload = new JSONObject();
|
||||||
|
payload.put("type", "gpio_out");
|
||||||
|
payload.put("body", body);
|
||||||
|
return payload;
|
||||||
|
}
|
||||||
|
|
||||||
|
public boolean set_io_lock_status(String sn, Integer status, Integer ioout) {
|
||||||
String operUrl = "device/%s/message/down/set_io_lock_status";
|
String operUrl = "device/%s/message/down/set_io_lock_status";
|
||||||
String topic = String.format(operUrl, sn);
|
String topic = String.format(operUrl, sn);
|
||||||
DeviceIOLockRequestPO deviceIOLockRequestPO = new DeviceIOLockRequestPO();
|
DeviceIOLockRequestPO deviceIOLockRequestPO = new DeviceIOLockRequestPO();
|
||||||
@@ -89,19 +98,7 @@ public class ServerService {
|
|||||||
deviceIOLockRequestPO.setSn(sn);
|
deviceIOLockRequestPO.setSn(sn);
|
||||||
deviceIOLockRequestPO.setName("set_io_lock_status");
|
deviceIOLockRequestPO.setName("set_io_lock_status");
|
||||||
deviceIOLockRequestPO.setPayload(this.buildPayloadLocked(status, ioout));
|
deviceIOLockRequestPO.setPayload(this.buildPayloadLocked(status, ioout));
|
||||||
boolean result = server.publish(sn,topic, JSON.toJSONString(deviceIOLockRequestPO).getBytes(StandardCharsets.UTF_8));
|
return server.publish(sn,topic, JSON.toJSONString(deviceIOLockRequestPO).getBytes(StandardCharsets.UTF_8));
|
||||||
return result;
|
|
||||||
}
|
|
||||||
|
|
||||||
private JSONObject buildPayloadOpen() {
|
|
||||||
JSONObject body = new JSONObject();
|
|
||||||
body.put("delay", 10000);
|
|
||||||
body.put("io", 0);
|
|
||||||
body.put("value",2);
|
|
||||||
JSONObject payload = new JSONObject();
|
|
||||||
payload.put("type", "gpio_out");
|
|
||||||
payload.put("body", body);
|
|
||||||
return payload;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private JSONObject buildPayloadLocked(Integer status, Integer ioout) {
|
private JSONObject buildPayloadLocked(Integer status, Integer ioout) {
|
||||||
@@ -117,52 +114,28 @@ public class ServerService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public boolean lockOpen(String sn) {
|
public boolean lockOpen(String sn) {
|
||||||
String operUrl = "device/%s/message/down/set_io_lock_status";
|
String operUrl = "device/%s/message/down/gpio_out";
|
||||||
String topic = String.format(operUrl, sn);
|
String topic = String.format(operUrl, sn);
|
||||||
|
OpenFloodgatePO openFloodgatePO = new OpenFloodgatePO();
|
||||||
//通道1解锁
|
String uuid = "open_" + UuidUtil.getUuid();
|
||||||
DeviceIOLockRequestPO deviceIOLockRequestPO = new DeviceIOLockRequestPO();
|
openFloodgatePO.setId(uuid);
|
||||||
String uuid = "lockOpen_" + UuidUtil.getUuid();
|
openFloodgatePO.setSn(sn);
|
||||||
deviceIOLockRequestPO.setId(uuid);
|
openFloodgatePO.setName("gpio_out");
|
||||||
deviceIOLockRequestPO.setSn(sn);
|
openFloodgatePO.setTimestamp(System.currentTimeMillis() / 1000);
|
||||||
deviceIOLockRequestPO.setName("set_io_lock_status");
|
openFloodgatePO.setPayload(this.buildGpioOut(0, 2));
|
||||||
deviceIOLockRequestPO.setPayload(this.buildPayloadLocked(0, 1));
|
return server.publish(sn,topic, JSON.toJSONString(openFloodgatePO).getBytes(StandardCharsets.UTF_8));
|
||||||
boolean result = server.publish(sn,topic, JSON.toJSONString(deviceIOLockRequestPO).getBytes(StandardCharsets.UTF_8));
|
|
||||||
|
|
||||||
//通道0 锁定
|
|
||||||
|
|
||||||
DeviceIOLockRequestPO po = new DeviceIOLockRequestPO();
|
|
||||||
po.setId("lockOpen_" + UuidUtil.getUuid());
|
|
||||||
po.setSn(sn);
|
|
||||||
po.setName("set_io_lock_status");
|
|
||||||
po.setPayload(this.buildPayloadLocked(1, 0));
|
|
||||||
boolean result2 = server.publish(sn,topic, JSON.toJSONString(po).getBytes(StandardCharsets.UTF_8));
|
|
||||||
|
|
||||||
return (result && result2);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean lockClose(String sn) {
|
public boolean lockClose(String sn) {
|
||||||
String operUrl = "device/%s/message/down/set_io_lock_status";
|
String operUrl = "device/%s/message/down/gpio_out";
|
||||||
String topic = String.format(operUrl, sn);
|
String topic = String.format(operUrl, sn);
|
||||||
|
OpenFloodgatePO openFloodgatePO = new OpenFloodgatePO();
|
||||||
//通道0解锁
|
String uuid = "open_" + UuidUtil.getUuid();
|
||||||
DeviceIOLockRequestPO deviceIOLockRequestPO = new DeviceIOLockRequestPO();
|
openFloodgatePO.setId(uuid);
|
||||||
String uuid = "lockOpen_" + UuidUtil.getUuid();
|
openFloodgatePO.setSn(sn);
|
||||||
deviceIOLockRequestPO.setId(uuid);
|
openFloodgatePO.setName("gpio_out");
|
||||||
deviceIOLockRequestPO.setSn(sn);
|
openFloodgatePO.setTimestamp(System.currentTimeMillis() / 1000);
|
||||||
deviceIOLockRequestPO.setName("set_io_lock_status");
|
openFloodgatePO.setPayload(this.buildGpioOut(1, 2));
|
||||||
deviceIOLockRequestPO.setPayload(this.buildPayloadLocked(0, 0));
|
return server.publish(sn,topic, JSON.toJSONString(openFloodgatePO).getBytes(StandardCharsets.UTF_8));
|
||||||
boolean result = server.publish(sn,topic, JSON.toJSONString(deviceIOLockRequestPO).getBytes(StandardCharsets.UTF_8));
|
|
||||||
|
|
||||||
//通道1 锁定
|
|
||||||
|
|
||||||
DeviceIOLockRequestPO po = new DeviceIOLockRequestPO();
|
|
||||||
po.setId("lockOpen_" + UuidUtil.getUuid());
|
|
||||||
po.setSn(sn);
|
|
||||||
po.setName("set_io_lock_status");
|
|
||||||
po.setPayload(this.buildPayloadLocked(1, 1));
|
|
||||||
boolean result2 = server.publish(sn,topic, JSON.toJSONString(po).getBytes(StandardCharsets.UTF_8));
|
|
||||||
|
|
||||||
return (result && result2);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -50,93 +50,98 @@ 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);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
//删除车牌
|
|
||||||
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));
|
|
||||||
|
|
||||||
//更新车辆授权信息
|
String topic = String.format(whiteUrl, car.getSn());
|
||||||
if (CollUtil.isNotEmpty(carParkRecords)) {
|
WhiteListOperatorPO po = new WhiteListOperatorPO();
|
||||||
carParkRecordService.saveOrUpdateBatch(carParkRecords);
|
String uuid = "add_" + UuidUtil.getUuid();
|
||||||
}
|
po.setId(uuid);
|
||||||
//发布车牌到设备
|
po.setSn(car.getSn());
|
||||||
boolean publish = mqttServer.publish(sn, topic, JSON.toJSONString(po).getBytes(StandardCharsets.UTF_8));
|
po.setName("white_list_operator");
|
||||||
log.info("定时任务删除车牌topic:{},报文数据:{},发送结果:{}", topic, JSON.toJSONString(po), publish);
|
//构造数据
|
||||||
|
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);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -1,12 +1,33 @@
|
|||||||
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
|
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
|
||||||
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
|
||||||
|
|||||||
@@ -2,13 +2,32 @@ spring:
|
|||||||
datasource:
|
datasource:
|
||||||
driver-class-name: com.mysql.cj.jdbc.Driver
|
driver-class-name: com.mysql.cj.jdbc.Driver
|
||||||
#td
|
#td
|
||||||
url: jdbc:mysql://192.168.251.16:3306/td_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8
|
url: jdbc:mysql://192.168.251.16:3306/td_cloud?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&connectTimeout=5000&socketTimeout=30000
|
||||||
username: root
|
username: root
|
||||||
password: td@JJ2024
|
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:
|
||||||
host: 192.168.251.16
|
host: 192.168.251.16
|
||||||
port: 6379
|
port: 6379
|
||||||
password:
|
password:
|
||||||
database: 2
|
database: 2
|
||||||
|
# Redis 超时配置,防止连接阻塞
|
||||||
|
timeout: 5000ms
|
||||||
|
lettuce:
|
||||||
|
pool:
|
||||||
|
max-active: 8
|
||||||
|
max-idle: 8
|
||||||
|
min-idle: 2
|
||||||
|
max-wait: 5000ms
|
||||||
|
|||||||
@@ -127,3 +127,8 @@ 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
|
||||||
|
|||||||
Reference in New Issue
Block a user