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

判断异常

server_five_liuyunhu
dongxiaodong 2024-04-05 11:37:29 +08:00
parent 976f35f212
commit 371c80d110
9 changed files with 347 additions and 344 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
couplet-common/couplet-common-system/src/main/java/com/couplet/common/system/remote/RemoteCodeService.java
View File

@ -17,6 +17,6 @@ import org.springframework.web.bind.annotation.RequestBody;
@FeignClient(contextId = "remoteCodeService",value = ServiceNameConstants.BUSINESS_SERVICE, fallbackFactory = RemoteCodeFallbackFactory.class) @FeignClient(contextId = "remoteCodeService",value = ServiceNameConstants.BUSINESS_SERVICE, fallbackFactory = RemoteCodeFallbackFactory.class)
public interface RemoteCodeService { public interface RemoteCodeService {
@PostMapping("insertCode") @PostMapping("trouble/insertCode")
public Result<Integer> insertCode(@RequestBody CoupletTroubleLog coupletTroubleLog); public Result<Integer> insertCode(@RequestBody CoupletTroubleLog coupletTroubleLog);
} }

28
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelMessage.java
View File

@ -48,16 +48,16 @@ public class ModelMessage {
} }
}; };
@Value("${mq.queueName}") // @Value("${mq.queueName}")
public String queueName; // public String queueName;
//
//交换机 // //交换机
@Value("${mq.exchangeName}") // @Value("${mq.exchangeName}")
public String exchangeName; // public String exchangeName;
//
//路由键 // //路由键
@Value("${mq.routingKey}") // @Value("${mq.routingKey}")
public String routingKey; // public String routingKey;
@Scheduled(cron = "0/5 * * * * ?") @Scheduled(cron = "0/5 * * * * ?")
public void startMsg() { public void startMsg() {
@ -85,10 +85,10 @@ public class ModelMessage {
for (CoupletMsgData msgData : coupletMsgDataList) { for (CoupletMsgData msgData : coupletMsgDataList) {
log.info("解析到车辆数据：{}", msgData); log.info("解析到车辆数据：{}", msgData);
//发送消息到MQ //发送消息到MQ
rabbitTemplate.convertAndSend("send-couplet-code",msgData,message -> { // rabbitTemplate.convertAndSend("send-couplet-code",msgData,message -> {
message.getMessageProperties().setMessageId(UUID.randomUUID().toString()); // message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
return message; // return message;
}); // });
for (String string : strings) { for (String string : strings) {
IncidentService incidentService = SpringUtils.getBean(string); IncidentService incidentService = SpringUtils.getBean(string);
incidentService.incident(msgData); incidentService.incident(msgData);

12
couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/bootstrap.yml
View File

@ -15,6 +15,7 @@ spring:
discovery: discovery:
# 服务注册地址 # 服务注册地址
server-addr: 121.89.211.230:8848 server-addr: 121.89.211.230:8848
namespace: a439ce3f-2c42-4b4c-9c4d-c8db49933c15
config: config:
# 配置中心地址 # 配置中心地址
server-addr: 121.89.211.230:8848 server-addr: 121.89.211.230:8848
@ -23,6 +24,7 @@ spring:
# 共享配置 # 共享配置
shared-configs: shared-configs:
- application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension} - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
namespace: a439ce3f-2c42-4b4c-9c4d-c8db49933c15
main: main:
allow-bean-definition-overriding: true allow-bean-definition-overriding: true
rabbitmq: rabbitmq:
@ -49,8 +51,8 @@ mybatis-plus:
configuration: configuration:
map-underscore-to-camel-case: true map-underscore-to-camel-case: true
# RabbitMQ配置 ## RabbitMQ配置
mq: #mq:
queueName: queue # queueName: queueName
exchangeName: exchange # exchangeName: exchangeName
routingKey: routingKey # routingKey: routingKey

194
couplet-modules/couplet-business/src/main/java/com/couplet/business/server/consumer/SendCodeQueueConsumer.java
View File

@ -1,97 +1,97 @@
package com.couplet.business.server.consumer; //package com.couplet.business.server.consumer;
//
import com.alibaba.fastjson.JSONObject; //import com.alibaba.fastjson.JSONObject;
import com.couplet.analyze.msg.domain.CoupletMsgData; //import com.couplet.analyze.msg.domain.CoupletMsgData;
import com.couplet.common.domain.CoupletTroubleLog; //import com.couplet.common.domain.CoupletTroubleLog;
import com.couplet.common.system.remote.RemoteCodeService; //import com.couplet.common.system.remote.RemoteCodeService;
import com.rabbitmq.client.Channel; //import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message; //import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Queue; //import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.RabbitListener; //import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate; //import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
import java.util.Date; //import java.util.Date;
import java.util.Random; //import java.util.Random;
import java.util.concurrent.CompletableFuture; //import java.util.concurrent.CompletableFuture;
//
/** ///**
* @author DongXiaoDong // * @author DongXiaoDong
* @version 1.0 // * @version 1.0
* @date 2024/3/14 22:09 // * @date 2024/3/14 22:09
* @description // * @description
*/ // */
@Component //@Component
@Slf4j //@Slf4j
public class SendCodeQueueConsumer { //public class SendCodeQueueConsumer {
@Autowired // @Autowired
private RedisTemplate<String, String> redisTemplate; // private RedisTemplate<String, String> redisTemplate;
//
@Autowired // @Autowired
private RemoteCodeService remoteCodeService; // private RemoteCodeService remoteCodeService;
//
@RabbitListener(queuesToDeclare = {@Queue("send-couplet-code")}) // @RabbitListener(queuesToDeclare = {@Queue("send-couplet-code")})
public void sendLogQueueConsumer(Message message, CoupletMsgData msgData, Channel channel) { // public void sendLogQueueConsumer(Message message, CoupletMsgData msgData, Channel channel) {
log.info("日志队列：{}，接收到的消息：{}，开始消费...","send-couplet-code", JSONObject.toJSONString(msgData)); // log.info("日志队列：{}，接收到的消息：{}，开始消费...","send-couplet-code", JSONObject.toJSONString(msgData));
long start = System.currentTimeMillis(); // long start = System.currentTimeMillis();
//
String messageId = message.getMessageProperties().getMessageId(); // String messageId = message.getMessageProperties().getMessageId();
//
try { // try {
Long aLong = redisTemplate.opsForSet().add("send-log-queue", messageId); // Long aLong = redisTemplate.opsForSet().add("send-log-queue", messageId);
if (aLong==1) { // if (aLong==1) {
//异步保存日志 // //异步保存日志
CompletableFuture.runAsync(() -> { // CompletableFuture.runAsync(() -> {
CoupletTroubleLog coupletTroubleLog = new CoupletTroubleLog(); // CoupletTroubleLog coupletTroubleLog = new CoupletTroubleLog();
//判断状态是否为异常 // //判断状态是否为异常
if (msgData.getVehicleStatus() !=1){ // if (msgData.getVehicleStatus() !=1){
String code = generateGTA(); // String code = generateGTA();
coupletTroubleLog.setTroubleLogCode(code); // coupletTroubleLog.setTroubleLogCode(code);
coupletTroubleLog.setTroubleLogStart(new Date()); // coupletTroubleLog.setTroubleLogStart(new Date());
String vin = msgData.getVin(); // String vin = msgData.getVin();
coupletTroubleLog.setTroubleLogVin(vin); // coupletTroubleLog.setTroubleLogVin(vin);
// 如果状态为正常1时添加结束时间 // // 如果状态为正常1时添加结束时间
if (msgData.getVehicleStatus() == 1){ // if (msgData.getVehicleStatus() == 1){
coupletTroubleLog.setTroubleLogEnd(new Date()); // coupletTroubleLog.setTroubleLogEnd(new Date());
} // }
} // }
remoteCodeService.insertCode(coupletTroubleLog); // remoteCodeService.insertCode(coupletTroubleLog);
}); // });
log.info(""); // log.info("记录异常成功");
} // }
long end = System.currentTimeMillis(); // long end = System.currentTimeMillis();
log.info("日志队列：{}，接收到的消息：{}，消费完成，耗时：{}毫秒","send-log-queue", JSONObject.toJSONString(msgData), (end-start)); // log.info("日志队列：{}，接收到的消息：{}，消费完成，耗时：{}毫秒","send-log-queue", JSONObject.toJSONString(msgData), (end-start));
} catch (Exception e) { // } catch (Exception e) {
throw new RuntimeException(e); // throw new RuntimeException(e);
} // }
} // }
/** // /**
* // *
* 拼接GTA字符串 // * 拼接GTA字符串
* @return // * @return
*/ // */
public static String generateGTA() { // public static String generateGTA() {
// 生成以GTA开头的字符串 // // 生成以GTA开头的字符串
String codefix = "GTA"; // String codefix = "GTA";
// 删除4位数随机数字 // // 删除4位数随机数字
String s = generateRandomNumber(4); // String s = generateRandomNumber(4);
//拼接 // //拼接
return codefix + s; // return codefix + s;
} // }
//
/** // /**
* 随机生成1到10位的数字 // * 随机生成1到10位的数字
* @param length // * @param length
* @return // * @return
*/ // */
public static String generateRandomNumber(int length) { // public static String generateRandomNumber(int length) {
Random random = new Random(); // Random random = new Random();
StringBuilder builder = new StringBuilder(); // StringBuilder builder = new StringBuilder();
for (int i = 0; i < length; i++) { // for (int i = 0; i < length; i++) {
builder.append(random.nextInt(10)); // builder.append(random.nextInt(10));
} // }
return builder.toString(); // return builder.toString();
} // }
} //}

3
couplet-modules/couplet-business/src/main/resources/bootstrap.yml
View File

@ -7,7 +7,6 @@ spring:
application: application:
# 应用名称 # 应用名称
name: couplet-business name: couplet-business
profiles: profiles:
# 环境配置 # 环境配置
active: dev active: dev
@ -16,6 +15,7 @@ spring:
discovery: discovery:
# 服务注册地址 # 服务注册地址
server-addr: 121.89.211.230:8848 server-addr: 121.89.211.230:8848
namespace: a439ce3f-2c42-4b4c-9c4d-c8db49933c15
config: config:
# 配置中心地址 # 配置中心地址
server-addr: 121.89.211.230:8848 server-addr: 121.89.211.230:8848
@ -24,6 +24,7 @@ spring:
# 共享配置 # 共享配置
shared-configs: shared-configs:
- application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension} - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
namespace: a439ce3f-2c42-4b4c-9c4d-c8db49933c15
main: main:
allow-bean-definition-overriding: true allow-bean-definition-overriding: true
logging: logging:

2
couplet-modules/couplet-business/src/main/resources/mapper/business/SysTroubleMapper.xml
View File

@ -20,7 +20,7 @@
LEFT JOIN couplet_trouble_type y on t.type_id= y.type_id LEFT JOIN couplet_trouble_type y on t.type_id= y.type_id
</sql> </sql>
<insert id="insertMsgResq"> <insert id="insertMsgResq">
insert into couplet_trouble_log(trouble_log_code,toruble_log_vin,trouble_log_start,trouble_log_end) insert into couplet_trouble_log(trouble_log_code,trouble_log_vin,trouble_log_start,trouble_log_end)
values(#{troubleLogCode},#{troubleLogVin},#{troubleLogStart},#{troubleLogEnd}) values(#{troubleLogCode},#{troubleLogVin},#{troubleLogStart},#{troubleLogEnd})
</insert> </insert>

12
couplet-modules/couplet-modules-mq/src/main/java/com/couplet/mq/config/RabbitMQConfig.java
View File

@ -25,16 +25,16 @@ import org.springframework.context.annotation.Primary;
public class RabbitMQConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback { public class RabbitMQConfig implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
// 通过注入的方式获取队列名、交换机名和路由键 // 通过注入的方式获取队列名、交换机名和路由键
//队列名 //队列名
@Value("${mq.queueName}") // @Value("${mq.queueName}")
public String queueName; public static final String queueName = "queueName";
//交换机 //交换机
@Value("${mq.exchangeName}") // @Value("${mq.exchangeName}")
public String exchangeName; public static final String exchangeName = "exchangeName";
//路由键 //路由键
@Value("${mq.routingKey}") // @Value("${mq.routingKey}")
public String routingKey; public static final String routingKey = "routingKey";
private RabbitTemplate rabbitTemplate; private RabbitTemplate rabbitTemplate;

110
couplet-modules/couplet-modules-mq/src/main/java/com/couplet/mq/controller/MqController.java
View File

@ -1,55 +1,55 @@
package com.couplet.mq.controller; //package com.couplet.mq.controller;
//
import com.couplet.common.core.utils.uuid.IdUtils; //import com.couplet.common.core.utils.uuid.IdUtils;
import com.couplet.mq.domain.User; //import com.couplet.mq.domain.User;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData; //import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate; //import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value; //import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.*; //import org.springframework.web.bind.annotation.*;
//
/** ///**
* @ProjectName: five-groups-couplet // * @ProjectName: five-groups-couplet
* @Author: LiuYunHu // * @Author: LiuYunHu
* @CreateTime: 2024/3/29 // * @CreateTime: 2024/3/29
* @Description: MQController类 // * @Description: MQController类
*/ // */
@RestController //@RestController
@RequestMapping("/mq") //@RequestMapping("/mq")
@Slf4j //@Slf4j
public class MqController { //public class MqController {
// 通过注入的方式获取队列名、交换机名和路由键 // // 通过注入的方式获取队列名、交换机名和路由键
//队列名 // //队列名
@Value("${mq.queueName}") //// @Value("${mq.queueName}")
public String queueName; // public String queueName;
//
//交换机 // //交换机
@Value("${mq.exchangeName}") //// @Value("${mq.exchangeName}")
public String exchangeName; // public String exchangeName;
//
//路由键 // //路由键
@Value("${mq.routingKey}") //// @Value("${mq.routingKey}")
public String routingKey; // public String routingKey;
//
@Autowired // @Autowired
private RabbitTemplate rabbitTemplate; // private RabbitTemplate rabbitTemplate;
//
/* // /*
* @Author: LiuYunHu // * @Author: LiuYunHu
* @Date: 2024/4/1 19:58 // * @Date: 2024/4/1 19:58
* @Description: 消息发送者 // * @Description: 消息发送者
* @Param: [data] // * @Param: [data]
* @Return: void // * @Return: void
**/ // **/
@PostMapping("/sout") // @PostMapping("/sout")
//接收json字符串 // //接收json字符串
public void sout(@RequestBody User param) { // public void sout(@RequestBody User param) {
rabbitTemplate.convertAndSend(exchangeName, routingKey, param, message -> { // rabbitTemplate.convertAndSend(exchangeName, routingKey, param, message -> {
message.getMessageProperties().setMessageId(IdUtils.randomUUID()); // message.getMessageProperties().setMessageId(IdUtils.randomUUID());
return message; // return message;
}, new CorrelationData(IdUtils.randomUUID()) // }, new CorrelationData(IdUtils.randomUUID())
); // );
//
} // }
} //}

328
couplet-modules/couplet-modules-mq/src/main/java/com/couplet/mq/service/Consumer.java
View File

@ -1,164 +1,164 @@
package com.couplet.mq.service; //package com.couplet.mq.service;
//
import com.couplet.mq.domain.User; //import com.couplet.mq.domain.User;
import com.rabbitmq.client.Channel; //import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message; //import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler; //import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener; //import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate; //import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component; //import org.springframework.stereotype.Component;
//
import java.io.IOException; //import java.io.IOException;
import java.util.concurrent.TimeUnit; //import java.util.concurrent.TimeUnit;
//
/** ///**
* @ProjectName: five-groups-couplet // * @ProjectName: five-groups-couplet
* @Author: LiuYunHu // * @Author: LiuYunHu
* @CreateTime: 2024/3/28 // * @CreateTime: 2024/3/28
* @Description: MQ消费者类 // * @Description: MQ消费者类
*/ // */
//
@Component //@Component
@Slf4j //@Slf4j
@SuppressWarnings("all") //@SuppressWarnings("all")
@RabbitListener(queues = "${mq.queueName}") //@RabbitListener(queues = "${mq.queueName}")
public class Consumer { //public class Consumer {
@Autowired // @Autowired
private StringRedisTemplate redis; // private StringRedisTemplate redis;
//
/* 线程池执行 // /* 线程池执行
//
//创建一个定长线程池 // //创建一个定长线程池
private final Executor executor = Executors.newFixedThreadPool(5); // private final Executor executor = Executors.newFixedThreadPool(5);
//
@Async // @Async
@RabbitHandler // @RabbitHandler
public void process(User param, Channel channel, Message message) { // public void process(User param, Channel channel, Message message) {
executor.execute(() -> { // executor.execute(() -> {
try { // try {
handleMessage(param, channel, message); // handleMessage(param, channel, message);
} catch (IOException e) { // } catch (IOException e) {
log.error("处理消息失败：{}", e); // log.error("处理消息失败：{}", e);
} // }
}); // });
} // }
//
//处理信息的方法 // //处理信息的方法
private void handleMessage(User param, Channel channel, Message message) throws IOException { // private void handleMessage(User param, Channel channel, Message message) throws IOException {
log.info("消费者收到消息为：{},{}" + param, message.getMessageProperties().getDeliveryTag()); // log.info("消费者收到消息为：{},{}" + param, message.getMessageProperties().getDeliveryTag());
//
long deliveryTag = message.getMessageProperties().getDeliveryTag(); // long deliveryTag = message.getMessageProperties().getDeliveryTag();
String messageId = message.getMessageProperties().getMessageId(); // String messageId = message.getMessageProperties().getMessageId();
//
if (!redis.hasKey("value:" + messageId)) { // if (!redis.hasKey("value:" + messageId)) {
redis.opsForValue().set("value:" + messageId, "" + deliveryTag, 5, TimeUnit.MINUTES); // redis.opsForValue().set("value:" + messageId, "" + deliveryTag, 5, TimeUnit.MINUTES);
} // }
//
// 1 添加成功新数据 0已有重复值,不允许再添加 // // 1 添加成功新数据 0已有重复值,不允许再添加
Long add = redis.opsForSet().add("set:" + messageId, "set:" + messageId); // Long add = redis.opsForSet().add("set:" + messageId, "set:" + messageId);
//过期时间 // //过期时间
redis.expire("set:" + messageId, 5, TimeUnit.MINUTES); // redis.expire("set:" + messageId, 5, TimeUnit.MINUTES);
//
//
try { // try {
if (add == 1) { // if (add == 1) {
//第一次 消费 // //第一次 消费
System.out.println("*****************************"); // System.out.println("*****************************");
System.out.println("消费者收到消息：" + param); // System.out.println("消费者收到消息：" + param);
System.out.println("*****************************"); // System.out.println("*****************************");
log.info("消费结束"); // log.info("消费结束");
//
channel.basicAck(deliveryTag, false); // channel.basicAck(deliveryTag, false);
//
} else { // } else {
//重复消费 // //重复消费
log.error("重复消费"); // log.error("重复消费");
channel.basicReject(deliveryTag, false); // channel.basicReject(deliveryTag, false);
//
//删除缓存 // //删除缓存
redis.opsForSet().remove("set:" + messageId, "set:" + messageId); // redis.opsForSet().remove("set:" + messageId, "set:" + messageId);
} // }
//
//
} catch (Exception e) { // } catch (Exception e) {
log.error("消息没有成功消费！"); // log.error("消息没有成功消费！");
//
String s = redis.opsForValue().get("value:" + messageId); // String s = redis.opsForValue().get("value:" + messageId);
//
long oldTag = Long.parseLong(s); // long oldTag = Long.parseLong(s);
//
if (deliveryTag == (oldTag + 2)) { // if (deliveryTag == (oldTag + 2)) {
log.error("确实消费不了，不入队了！"); // log.error("确实消费不了，不入队了！");
channel.basicNack(deliveryTag, false, false); // channel.basicNack(deliveryTag, false, false);
} else { // } else {
log.info("消息消费失败，重新入队"); // log.info("消息消费失败，重新入队");
channel.basicNack(deliveryTag, false, true); // channel.basicNack(deliveryTag, false, true);
} // }
} // }
//
} // }
//
**/ //**/
//
@RabbitHandler // @RabbitHandler
public void process(User param, Channel channel, Message message) throws IOException { // public void process(User param, Channel channel, Message message) throws IOException {
log.info("消费者收到消息为：{},{}" + param, message.getMessageProperties().getDeliveryTag()); // log.info("消费者收到消息为：{},{}" + param, message.getMessageProperties().getDeliveryTag());
//
long deliveryTag = message.getMessageProperties().getDeliveryTag(); // long deliveryTag = message.getMessageProperties().getDeliveryTag();
String messageId = message.getMessageProperties().getMessageId(); // String messageId = message.getMessageProperties().getMessageId();
//
if (!redis.hasKey("value:" + messageId)) { // if (!redis.hasKey("value:" + messageId)) {
redis.opsForValue().set("value:" + messageId, "" + deliveryTag, 5, TimeUnit.MINUTES); // redis.opsForValue().set("value:" + messageId, "" + deliveryTag, 5, TimeUnit.MINUTES);
} // }
//
// 1 添加成功新数据 0已有重复值,不允许再添加 // // 1 添加成功新数据 0已有重复值,不允许再添加
Long add = redis.opsForSet().add("set:" + messageId, "set:" + messageId); // Long add = redis.opsForSet().add("set:" + messageId, "set:" + messageId);
//过期时间 // //过期时间
redis.expire("set:" + messageId, 5, TimeUnit.MINUTES); // redis.expire("set:" + messageId, 5, TimeUnit.MINUTES);
//
//
try { // try {
if (add == 1) { // if (add == 1) {
//第一次 消费 // //第一次 消费
System.out.println("*****************************"); // System.out.println("*****************************");
System.out.println("消费者收到消息：" + param); // System.out.println("消费者收到消息：" + param);
System.out.println("*****************************"); // System.out.println("*****************************");
log.info("消费结束"); // log.info("消费结束");
//
//确认消费 // //确认消费
channel.basicAck(deliveryTag, false); // channel.basicAck(deliveryTag, false);
//
} else { // } else {
//重复消费 // //重复消费
log.error("重复消费"); // log.error("重复消费");
//拒绝消费 // //拒绝消费
channel.basicReject(deliveryTag, false); // channel.basicReject(deliveryTag, false);
//
//删除缓存 // //删除缓存
redis.opsForSet().remove("set:" + messageId, "set:" + messageId); // redis.opsForSet().remove("set:" + messageId, "set:" + messageId);
} // }
//
//
} catch (Exception e) { // } catch (Exception e) {
log.error("消息没有成功消费！"); // log.error("消息没有成功消费！");
//
String s = redis.opsForValue().get("value:" + messageId); // String s = redis.opsForValue().get("value:" + messageId);
//
long oldTag = Long.parseLong(s); // long oldTag = Long.parseLong(s);
//
if (deliveryTag == (oldTag + 2)) { // if (deliveryTag == (oldTag + 2)) {
log.error("确实消费不了，不入队了！"); // log.error("确实消费不了，不入队了！");
//
//
//拒绝消费 // //拒绝消费
channel.basicNack(deliveryTag, false, false); // channel.basicNack(deliveryTag, false, false);
} else { // } else {
log.info("消息消费失败，重新入队"); // log.info("消息消费失败，重新入队");
//重新入队 // //重新入队
channel.basicNack(deliveryTag, false, true); // channel.basicNack(deliveryTag, false, true);
} // }
} // }
} // }
} //}