five-groups
/
five-groups-couplet
4
0
Fork 0
Code Issues Pull Requests 2 Packages Projects Releases Wiki Activity

feat: 电子围栏报警

server_five_liuyunhu
lijiayao 2024-04-07 22:37:13 +08:00
parent 8bbaadea94
commit 16d72968d4
4 changed files with 246 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
couplet-modules/couplet-analyze/couplet-analyze-msg/pom.xml
View File

@ -92,6 +92,10 @@
<groupId>com.couplet</groupId>
<artifactId>couplet-common-event</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
</dependencies>

166
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/consumer/FenceConsumer.java
View File

@ -1,84 +1,84 @@
package com.couplet.analyze.msg.consumer;
import com.couplet.common.core.text.Convert;
import com.couplet.common.domain.request.FenceUpdateRequest;
import com.couplet.common.redis.service.RedisService;
import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
* @Author: LiJiaYao
* @Date: 2024/4/4
* @Description:
*/
@Log4j2
@Component
@RabbitListener(queues = "fenceQueue")
public class FenceConsumer {
@Autowired
private RedisService redisService;
@RabbitHandler
public void fenceConsumer(FenceUpdateRequest fenceUpdateRequest, Channel channel, Message message) throws IOException {
log.info("电子围栏消息进入队列，传入的数据是:[{}]", fenceUpdateRequest);
String messageId = message.getMessageProperties().getMessageId();
long deliveryTag = message.getMessageProperties().getDeliveryTag();
if (!redisService.hasKey("电子围栏消息不丢失:" + messageId)) {
redisService.setCacheObject("电子围栏消息不丢失:" + messageId, "" + deliveryTag);
}
// if (redisService.hasKey("fence")){
// redisService.deleteObject("fence");
//package com.couplet.analyze.msg.consumer;
//
//import com.couplet.common.core.text.Convert;
//import com.couplet.common.domain.request.FenceUpdateRequest;
//import com.couplet.common.redis.service.RedisService;
//import com.rabbitmq.client.Channel;
//import lombok.extern.log4j.Log4j2;
//import org.springframework.amqp.core.Message;
//import org.springframework.amqp.rabbit.annotation.RabbitHandler;
//import org.springframework.amqp.rabbit.annotation.RabbitListener;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.core.BoundSetOperations;
//import org.springframework.stereotype.Component;
//
//import java.io.IOException;
//import java.util.HashMap;
//import java.util.HashSet;
//import java.util.Set;
//import java.util.concurrent.TimeUnit;
//
///**
// * @Author: LiJiaYao
// * @Date: 2024/4/4
// * @Description:
// */
//@Log4j2
//@Component
//@RabbitListener(queues = "fenceQueue")
//public class FenceConsumer {
// @Autowired
// private RedisService redisService;
//
// @RabbitHandler
// public void fenceConsumer(FenceUpdateRequest fenceUpdateRequest, Channel channel, Message message) throws IOException {
//
// log.info("电子围栏消息进入队列，传入的数据是:[{}]", fenceUpdateRequest);
//
// String messageId = message.getMessageProperties().getMessageId();
// long deliveryTag = message.getMessageProperties().getDeliveryTag();
// if (!redisService.hasKey("电子围栏消息不丢失:" + messageId)) {
// redisService.setCacheObject("电子围栏消息不丢失:" + messageId, "" + deliveryTag);
// }
//// if (redisService.hasKey("fence")){
//// redisService.deleteObject("fence");
//// }
//
// HashSet<String> objects = new HashSet<>();
// objects.add(messageId);
//
// BoundSetOperations<String, String> set = redisService.setCacheSet("电子围栏消息不重复:" + messageId, objects);
// redisService.expire("电子围栏消息不重复:" + messageId, 5, TimeUnit.MINUTES);
// try {
// if (set != null) {
// HashMap<String, Object> hashMap = new HashMap<>();
// HashSet<FenceUpdateRequest> hashSet = new HashSet<>();
// hashSet.add(fenceUpdateRequest);
// hashMap.put(fenceUpdateRequest.getFenceId()+"",fenceUpdateRequest);
//// redisTemplate.opsForH("fence", JSON.toJSONString(hashMap),10,TimeUnit.MINUTES);
//// redisTemplate.opsForHash().put("fence", fenceUpdateRequest.getFenceId()+"", JSON.toJSONString(hashMap));
//
// String key = Convert.toStr(fenceUpdateRequest.getFenceId());
// redisService.setCacheObject(key,fenceUpdateRequest);
// redisService.expire(key, 10, TimeUnit.MINUTES);
// //判断车辆是否有实时数据,如果没有则删除数据
// channel.basicAck(deliveryTag, false);
// } else {
// log.error("电子围栏消息不能重复消费:[{}]", fenceUpdateRequest);
// channel.basicReject(deliveryTag, false);
// }
// } catch (IOException e) {
// log.error("电子围栏消息未进入队列,传入的信息是:【{}】", fenceUpdateRequest);
// String s = redisService.getCacheObject("电子围栏消息不丢失:" + messageId);
//
// Long o = Long.valueOf(s);
// if (deliveryTag == o + 2) {
// log.error("电子围栏消息已丢失,无法传入的信息是:【{}】", fenceUpdateRequest);
// channel.basicNack(deliveryTag, false, false);
// } else {
// log.error("电子围栏消息已丢失,已再次传入的信息是:【{}】", fenceUpdateRequest);
// channel.basicNack(deliveryTag, true, false);
// }
// }
// }
//}
HashSet<String> objects = new HashSet<>();
objects.add(messageId);
BoundSetOperations<String, String> set = redisService.setCacheSet("电子围栏消息不重复:" + messageId, objects);
redisService.expire("电子围栏消息不重复:" + messageId, 5, TimeUnit.MINUTES);
try {
if (set != null) {
HashMap<String, Object> hashMap = new HashMap<>();
HashSet<FenceUpdateRequest> hashSet = new HashSet<>();
hashSet.add(fenceUpdateRequest);
hashMap.put(fenceUpdateRequest.getFenceId()+"",fenceUpdateRequest);
// redisTemplate.opsForH("fence", JSON.toJSONString(hashMap),10,TimeUnit.MINUTES);
// redisTemplate.opsForHash().put("fence", fenceUpdateRequest.getFenceId()+"", JSON.toJSONString(hashMap));
String key = Convert.toStr(fenceUpdateRequest.getFenceId());
redisService.setCacheObject(key,fenceUpdateRequest);
redisService.expire(key, 10, TimeUnit.MINUTES);
//判断车辆是否有实时数据,如果没有则删除数据
channel.basicAck(deliveryTag, false);
} else {
log.error("电子围栏消息不能重复消费:[{}]", fenceUpdateRequest);
channel.basicReject(deliveryTag, false);
}
} catch (IOException e) {
log.error("电子围栏消息未进入队列,传入的信息是:【{}】", fenceUpdateRequest);
String s = redisService.getCacheObject("电子围栏消息不丢失:" + messageId);
Long o = Long.valueOf(s);
if (deliveryTag == o + 2) {
log.error("电子围栏消息已丢失,无法传入的信息是:【{}】", fenceUpdateRequest);
channel.basicNack(deliveryTag, false, false);
} else {
log.error("电子围栏消息已丢失,已再次传入的信息是:【{}】", fenceUpdateRequest);
channel.basicNack(deliveryTag, true, false);
}
}
}
}

182
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/consumer/MsgConsumer.java
View File

@ -1,91 +1,91 @@
package com.couplet.analyze.msg.consumer;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.mapper.IncidentMapper;
import com.couplet.analyze.msg.service.impl.realTimeData.RealTimeJudge;
import com.couplet.common.domain.request.RealTimeDataRequest;
import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
/**
* @Author: LiJiaYao
* @Date: 2024/4/4
* @Description:
*/
@Log4j2
@Component
@RabbitListener(queues = "finByVinQueueName")
public class MsgConsumer {
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private IncidentMapper incidentMapper;
@RabbitHandler
public void realTimeDataConsumer(RealTimeDataRequest realTimeDataRequest, Channel channel, Message message) throws IOException {
log.info("消息进入队列，传入的数据是:[{}]", realTimeDataRequest);
String messageId = message.getMessageProperties().getMessageId();
long deliveryTag = message.getMessageProperties().getDeliveryTag();
if (!redisTemplate.hasKey("消息不丢失:" + messageId)) {
redisTemplate.opsForValue().set("消息不丢失:" + messageId, "" + deliveryTag, 1, TimeUnit.MINUTES);
}
Long add = redisTemplate.opsForSet().add("消息不重复:" + messageId, messageId);
redisTemplate.expire("消息不重复:" + messageId, 5, TimeUnit.MINUTES);
try {
if (0 < add) {
JSONObject jsonObject = JSONObject.parseObject(String.valueOf(realTimeDataRequest));
Long userId = jsonObject.getLong("userId");
String vin = jsonObject.getString("vin");
RealTimeDataRequest request = new RealTimeDataRequest();
request.setVin(vin);
request.setUserId(userId);
RealTimeJudge.addRealTime(request);
//判断车辆是否有实时数据,如果没有则删除数据
if (RealTimeJudge.isJudge(realTimeDataRequest.getVin())){
log.info("开始实时数据传输:[{}]",realTimeDataRequest.getVin());
}
CoupletMsgData incident = incidentMapper.queryByIncident(realTimeDataRequest.getVin());
if (incident == null){
log.error("没有数据......");
}
redisTemplate.opsForList().rightPush("coupletMsgData", JSON.toJSONString(incident));
channel.basicAck(deliveryTag, false);
} else {
log.error("消息不能重复消费:[{}]", realTimeDataRequest);
channel.basicReject(deliveryTag, false);
}
} catch (IOException e) {
log.error("消息未进入队列,传入的信息是:【{}】", realTimeDataRequest);
String s = redisTemplate.opsForValue().get("消息不丢失:" + messageId);
Long o = Long.valueOf(s);
if (deliveryTag == o + 2) {
log.error("消息已丢失,无法传入的信息是:【{}】", realTimeDataRequest);
channel.basicNack(deliveryTag, false, false);
} else {
log.error("消息已丢失,已再次传入的信息是:【{}】", realTimeDataRequest);
channel.basicNack(deliveryTag, true, false);
}
}
}
}
//package com.couplet.analyze.msg.consumer;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.couplet.analyze.msg.domain.CoupletMsgData;
//import com.couplet.analyze.msg.mapper.IncidentMapper;
//import com.couplet.analyze.msg.service.impl.realTimeData.RealTimeJudge;
//import com.couplet.common.domain.request.RealTimeDataRequest;
//import com.rabbitmq.client.Channel;
//import lombok.extern.log4j.Log4j2;
//import org.springframework.amqp.core.Message;
//import org.springframework.amqp.rabbit.annotation.RabbitHandler;
//import org.springframework.amqp.rabbit.annotation.RabbitListener;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.core.StringRedisTemplate;
//import org.springframework.stereotype.Component;
//
//import java.io.IOException;
//import java.util.concurrent.TimeUnit;
//
///**
// * @Author: LiJiaYao
// * @Date: 2024/4/4
// * @Description:
// */
//@Log4j2
//@Component
//@RabbitListener(queues = "finByVinQueueName")
//public class MsgConsumer {
// @Autowired
// private StringRedisTemplate redisTemplate;
// @Autowired
// private IncidentMapper incidentMapper;
//
// @RabbitHandler
// public void realTimeDataConsumer(RealTimeDataRequest realTimeDataRequest, Channel channel, Message message) throws IOException {
//
// log.info("消息进入队列，传入的数据是:[{}]", realTimeDataRequest);
//
// String messageId = message.getMessageProperties().getMessageId();
// long deliveryTag = message.getMessageProperties().getDeliveryTag();
// if (!redisTemplate.hasKey("消息不丢失:" + messageId)) {
// redisTemplate.opsForValue().set("消息不丢失:" + messageId, "" + deliveryTag, 1, TimeUnit.MINUTES);
// }
//
// Long add = redisTemplate.opsForSet().add("消息不重复:" + messageId, messageId);
// redisTemplate.expire("消息不重复:" + messageId, 5, TimeUnit.MINUTES);
// try {
// if (0 < add) {
// JSONObject jsonObject = JSONObject.parseObject(String.valueOf(realTimeDataRequest));
// Long userId = jsonObject.getLong("userId");
// String vin = jsonObject.getString("vin");
// RealTimeDataRequest request = new RealTimeDataRequest();
// request.setVin(vin);
// request.setUserId(userId);
// RealTimeJudge.addRealTime(request);
// //判断车辆是否有实时数据,如果没有则删除数据
// if (RealTimeJudge.isJudge(realTimeDataRequest.getVin())){
// log.info("开始实时数据传输:[{}]",realTimeDataRequest.getVin());
// }
// CoupletMsgData incident = incidentMapper.queryByIncident(realTimeDataRequest.getVin());
// if (incident == null){
// log.error("没有数据......");
// }
// redisTemplate.opsForList().rightPush("coupletMsgData", JSON.toJSONString(incident));
//
// channel.basicAck(deliveryTag, false);
// } else {
// log.error("消息不能重复消费:[{}]", realTimeDataRequest);
// channel.basicReject(deliveryTag, false);
// }
// } catch (IOException e) {
//
// log.error("消息未进入队列,传入的信息是:【{}】", realTimeDataRequest);
// String s = redisTemplate.opsForValue().get("消息不丢失:" + messageId);
//
// Long o = Long.valueOf(s);
// if (deliveryTag == o + 2) {
// log.error("消息已丢失,无法传入的信息是:【{}】", realTimeDataRequest);
// channel.basicNack(deliveryTag, false, false);
// } else {
// log.error("消息已丢失,已再次传入的信息是:【{}】", realTimeDataRequest);
// channel.basicNack(deliveryTag, true, false);
// }
//
// }
//
//
// }
//
//}

136
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/consumer/VehicleConsumer.java
View File

@ -1,68 +1,68 @@
package com.couplet.analyze.msg.consumer;
import com.couplet.common.core.text.Convert;
import com.couplet.common.redis.service.RedisService;
import com.rabbitmq.client.Channel;
import lombok.extern.log4j.Log4j2;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* @Author: LiJiaYao
* @Date: 2024/4/4
* @Description:
*/
@Log4j2
@Component
@RabbitListener(queues = "vehicleQueue")
public class VehicleConsumer {
@Autowired
private RedisService redisService;
@RabbitHandler
public void vehicleConsumer(String vehicleAndLogo, Channel channel, Message message) throws IOException {
log.info("车辆消息进入队列，传入的数据是:[{}]", vehicleAndLogo);
String messageId = message.getMessageProperties().getMessageId();
long deliveryTag = message.getMessageProperties().getDeliveryTag();
if (!redisService.hasKey("车辆消息不丢失:" + messageId)) {
redisService.setCacheObject("车辆消息不丢失:" + messageId, "" + deliveryTag);
}
HashSet<String> objects = new HashSet<>();
objects.add(messageId);
BoundSetOperations<String, String> set = redisService.setCacheSet("车辆信息消息不重复:" + messageId, objects);
redisService.expire("车辆信息消息不重复:" + messageId, 5, TimeUnit.MINUTES);
try {
if (set != null) {
// String key = Convert.toStr(id);
String key = "id";
redisService.setCacheObject(key, vehicleAndLogo);
redisService.expire(key, 10, TimeUnit.MINUTES);
//判断车辆是否有实时数据,如果没有则删除数据
channel.basicAck(deliveryTag, false);
} else {
log.error("车辆消息不能重复消费:[{}]", vehicleAndLogo);
channel.basicReject(deliveryTag, false);
}
} catch (IOException e) {
log.error("车辆消息未进入队列,传入的信息是:【{}】", vehicleAndLogo);
String s = redisService.getCacheObject("车辆消息不丢失:" + messageId);
Long o = Long.valueOf(s);
if (deliveryTag == o + 2) {
log.error("车辆消息已丢失,无法传入的信息是:【{}】", vehicleAndLogo);
channel.basicNack(deliveryTag, false, false);
} else {
log.error("车辆消息已丢失,已再次传入的信息是:【{}】", vehicleAndLogo);
channel.basicNack(deliveryTag, true, false);
}
}
}
}
//package com.couplet.analyze.msg.consumer;
//
//import com.couplet.common.core.text.Convert;
//import com.couplet.common.redis.service.RedisService;
//import com.rabbitmq.client.Channel;
//import lombok.extern.log4j.Log4j2;
//import org.springframework.amqp.core.Message;
//import org.springframework.amqp.rabbit.annotation.RabbitHandler;
//import org.springframework.amqp.rabbit.annotation.RabbitListener;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.data.redis.core.BoundSetOperations;
//import org.springframework.stereotype.Component;
//
//import java.io.IOException;
//import java.util.HashSet;
//import java.util.List;
//import java.util.concurrent.TimeUnit;
//
///**
// * @Author: LiJiaYao
// * @Date: 2024/4/4
// * @Description:
// */
//@Log4j2
//@Component
//@RabbitListener(queues = "vehicleQueue")
//public class VehicleConsumer {
// @Autowired
// private RedisService redisService;
// @RabbitHandler
// public void vehicleConsumer(String vehicleAndLogo, Channel channel, Message message) throws IOException {
// log.info("车辆消息进入队列，传入的数据是:[{}]", vehicleAndLogo);
// String messageId = message.getMessageProperties().getMessageId();
// long deliveryTag = message.getMessageProperties().getDeliveryTag();
// if (!redisService.hasKey("车辆消息不丢失:" + messageId)) {
// redisService.setCacheObject("车辆消息不丢失:" + messageId, "" + deliveryTag);
// }
// HashSet<String> objects = new HashSet<>();
// objects.add(messageId);
// BoundSetOperations<String, String> set = redisService.setCacheSet("车辆信息消息不重复:" + messageId, objects);
// redisService.expire("车辆信息消息不重复:" + messageId, 5, TimeUnit.MINUTES);
// try {
// if (set != null) {
//// String key = Convert.toStr(id);
//
// String key = "id";
// redisService.setCacheObject(key, vehicleAndLogo);
// redisService.expire(key, 10, TimeUnit.MINUTES);
// //判断车辆是否有实时数据,如果没有则删除数据
// channel.basicAck(deliveryTag, false);
// } else {
// log.error("车辆消息不能重复消费:[{}]", vehicleAndLogo);
// channel.basicReject(deliveryTag, false);
// }
// } catch (IOException e) {
// log.error("车辆消息未进入队列,传入的信息是:【{}】", vehicleAndLogo);
// String s = redisService.getCacheObject("车辆消息不丢失:" + messageId);
// Long o = Long.valueOf(s);
// if (deliveryTag == o + 2) {
// log.error("车辆消息已丢失,无法传入的信息是:【{}】", vehicleAndLogo);
// channel.basicNack(deliveryTag, false, false);
// } else {
// log.error("车辆消息已丢失,已再次传入的信息是:【{}】", vehicleAndLogo);
// channel.basicNack(deliveryTag, true, false);
// }
// }
// }
//}