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

fix: 代码丢失找回

server_five_liuyunhu
lijiayao 2024-04-06 10:15:12 +08:00
parent e8c0690352
commit d2a29b0016
10 changed files with 214 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
couplet-common/couplet-common-business/src/main/java/com/couplet/remote/RemoteTroubleService.java 100644
View File

@ -0,0 +1,25 @@
package com.couplet.remote;
import com.couplet.common.core.constant.ServiceNameConstants;
import com.couplet.common.core.domain.Result;
import com.couplet.common.domain.CoupletTroubleCode;
import com.couplet.remote.factory.RemoteTroubleFallbackFactory;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
@FeignClient(contextId = "remoteTroubleService" ,
value = ServiceNameConstants.BUSINESS_SERVICE,
fallbackFactory = RemoteTroubleFallbackFactory.class
)
public interface RemoteTroubleService {
/**
* 新增故障码
* @param code
* @return
*/
@PostMapping("/trouble/newFaultData")
public Result<?> newFaultData(@RequestBody CoupletTroubleCode code);
}

3
couplet-common/couplet-common-business/src/main/java/com/couplet/remote/RemoteVehicleService.java
View File

@ -51,7 +51,4 @@ public interface RemoteVehicleService {
@GetMapping("/findByVIN/{vin}")
public Result<List<Vehicle>> findByVIN(@PathVariable("vin") String vin);
@GetMapping("onOrOutLineByVIN")
public Integer onOrOutLineByVIN(@RequestParam("params") String params);
}

34
couplet-common/couplet-common-business/src/main/java/com/couplet/remote/factory/RemoteTroubleFallbackFactory.java 100644
View File

@ -0,0 +1,34 @@
package com.couplet.remote.factory;
import com.couplet.common.core.domain.Result;
import com.couplet.common.domain.CoupletTroubleCode;
import com.couplet.remote.RemoteTroubleService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.openfeign.FallbackFactory;
import org.springframework.stereotype.Component;
/**
* @author fufanrui
* @version 1.0
* @description: TODO
* @date 2024/4/2 14:46
*/
@Component
public class RemoteTroubleFallbackFactory implements FallbackFactory<RemoteTroubleService> {
private static final Logger log = LoggerFactory.getLogger(RemoteTroubleFallbackFactory.class);
@Override
public RemoteTroubleService create(Throwable cause) {
return new RemoteTroubleService() {
@Override
public Result<?> newFaultData(CoupletTroubleCode code) {
return Result.error("调用失败...."+cause.getMessage());
}
};
}
}

6
couplet-common/couplet-common-business/src/main/java/com/couplet/remote/factory/RemoteVehicleFallbackFactory.java
View File

@ -50,12 +50,6 @@ public class RemoteVehicleFallbackFactory implements FallbackFactory<RemoteVehic
public Result<List<Vehicle>> findByVIN(String vin) {
return Result.error("车辆服务调用失败:" + cause.getMessage());
}
@Override
public Integer onOrOutLineByVIN(String params) {
log.error("车辆服务调用失败:"+cause.getMessage());
return null;
}
};
}
}

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

@ -0,0 +1,77 @@
package com.couplet.analyze.msg.consumer;
import com.alibaba.fastjson.JSON;
import com.couplet.common.domain.request.FenceUpdateRequest;
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.HashMap;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
/**
* @Author: LiJiaYao
* @Date: 2024/4/4
* @Description:
*/
@Log4j2
@Component
@RabbitListener(queues = "fenceQueue")
public class FenceConsumer {
@Autowired
private StringRedisTemplate redisTemplate;
@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 (!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) {
HashMap<String, Object> hashMap = new HashMap<>();
HashSet<FenceUpdateRequest> hashSet = new HashSet<>();
hashSet.add(fenceUpdateRequest);
hashMap.put(fenceUpdateRequest.getFenceId()+"",fenceUpdateRequest);
redisTemplate.opsForValue().set("fence", JSON.toJSONString(hashMap),10,TimeUnit.MINUTES);
//判断车辆是否有实时数据,如果没有则删除数据
channel.basicAck(deliveryTag, false);
} else {
log.error("消息不能重复消费:[{}]", fenceUpdateRequest);
channel.basicReject(deliveryTag, false);
}
} catch (IOException e) {
log.error("消息未进入队列,传入的信息是:【{}】", fenceUpdateRequest);
String s = redisTemplate.opsForValue().get("消息不丢失:" + 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);
}
}
}
}

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

@ -7,20 +7,15 @@ 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.*;
import org.springframework.amqp.core.Queue;
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.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
@ -30,28 +25,8 @@ import java.util.concurrent.TimeUnit;
*/
@Log4j2
@Component
//@RabbitListener(queues = "vinQueue")
@RabbitListener(queues = "finByVinQueueName")
public class MsgConsumer {
// 普通交换机名称
public static final String EXCHANGE_NAME = "confirm_exchange";
// 队列名称
public static final String QUEUE_NAME = "vinQueue";
public static final String ROUTING_KEY = "key1";
@Bean
public DirectExchange confirmExchange() {
return new DirectExchange(EXCHANGE_NAME);
}
@Bean
public Queue confirmQueue() {
return QueueBuilder.durable(QUEUE_NAME).build();
}
@Bean
public Binding queueBindingExchange(@Qualifier("confirmQueue") Queue confirmQueue, @Qualifier("confirmExchange") DirectExchange confirmExchange) {
return BindingBuilder.bind(confirmQueue).to(confirmExchange).with(ROUTING_KEY);
}
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired

52
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/BreakdownServiceImpl.java
View File

@ -4,17 +4,18 @@ import com.alibaba.fastjson.JSON;
import com.couplet.analyze.msg.contents.StateConstant;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService;
import lombok.extern.log4j.Log4j2;
import org.aspectj.bridge.Message;
import com.couplet.common.domain.CoupletTroubleCode;
import com.couplet.remote.RemoteTroubleService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.stereotype.Service;
import java.util.Date;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
@ -30,6 +31,9 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
*/
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private RemoteTroubleService remoteTroubleService;
private static Logger log = LoggerFactory.getLogger(BreakdownServiceImpl.class);
@ -66,17 +70,22 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
|| StateConstant.DCDC_STATUS != coupletMsgData.getDcdcStatus()
|| StateConstant.CHG_STATUS != coupletMsgData.getChgStatus()){
//获取过期的key
String expireKey = coupletMsgData.toString();
redisTemplate.opsForValue().set(String.valueOf(coupletMsgData),JSON.toJSONString(coupletMsgData),10, TimeUnit.MINUTES);
log.debug("失效+key is:"+ expireKey);
String key = "breakdown";
log.debug("失效+key is:"+ key);
String value = JSON.toJSONString(coupletMsgData);
redisTemplate.opsForSet().add(key, value);
long expireTime = 30;
redisTemplate.expire(key, expireTime, TimeUnit.MINUTES);
long timeMillis = System.currentTimeMillis();
scheduledRedis();
log.info("故障事件结束时间："+timeMillis);
log.info("故障事件检测结束.....");
log.info("故障事件结束.....");
}
long timeMillis = System.currentTimeMillis();
log.info("故障事件结束时间："+timeMillis);
log.info("故障事件检测结束.....");
log.info("故障事件结束.....");
}
/**
@ -86,4 +95,31 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
public String getName() {
return "breakdown";
}
public void scheduledRedis() {
// Get all members of the set
Set<String> members = redisTemplate.opsForSet().members("breakdown");
if (members.size()!=0){
for (String member : members){
CoupletMsgData code = JSON.parseObject(member, CoupletMsgData.class);
Set<String> breakdownIds = redisTemplate.opsForSet().members(code.getVin());
if (breakdownIds.size()==0){
CoupletTroubleCode troubleCode = new CoupletTroubleCode();
troubleCode.setTroubleStartTime(new Date());
// 插入数据库
troubleCode.setTroubleTag(0);
troubleCode.setTroubleVin(code.getVin());
remoteTroubleService.newFaultData(troubleCode);
redisTemplate.opsForSet().add(code.getVin(), code.getVin()+":"+code);
long expireTime = 30;
redisTemplate.expire(code.getVin(), expireTime, TimeUnit.MINUTES);
}
}
}
}
}

14
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/ElectronicFenceServiceImpl.java
View File

@ -3,6 +3,8 @@ package com.couplet.analyze.msg.service.impl;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.service.IncidentService;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
/**
@ -14,6 +16,8 @@ import org.springframework.stereotype.Service;
@Log4j2
public class ElectronicFenceServiceImpl implements IncidentService {
@Autowired
private StringRedisTemplate redisTemplate;
/**
* 电子围栏事件
@ -24,10 +28,16 @@ public class ElectronicFenceServiceImpl implements IncidentService {
public void incident(CoupletMsgData coupletMsgData) {
log.info("电子围栏事件开始.......");
if (redisTemplate.hasKey("fence")) {
log.info("电子围栏事件redis存在.......");
String s = redisTemplate.opsForValue().get("fence");
log.info("更改的电子围栏内容是:"+s);
log.info("电子围栏事件结束.......");
}
log.info("电子围栏事件结束.......");
}
/**

8
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/RealTimeDataServiceImpl.java
View File

@ -1,19 +1,14 @@
package com.couplet.analyze.msg.service.impl;
import com.alibaba.fastjson.JSON;
import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.analyze.msg.mapper.IncidentMapper;
import com.couplet.analyze.msg.service.IncidentService;
import com.couplet.analyze.msg.service.impl.realTimeData.RealTimeJudge;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
/**
* @Author: LiJiaYao
* @Date: 2024/4/2
@ -47,7 +42,6 @@ public class RealTimeDataServiceImpl implements IncidentService {
}
log.info("[{}]开始传输实时数据", coupletMsgData.getVin());
log.info("实时数据事件结束.....");
}

19
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/breakdown/BreakdownEvent.java 100644
View File

@ -0,0 +1,19 @@
package com.couplet.analyze.msg.service.impl.breakdown;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
/**
* @Author: LiJiaYao
* @Date: 2024/4/5
* @Description:
*/
@Component
public class BreakdownEvent {
@Autowired
private StringRedisTemplate redisTemplate;
}