admin/vlv-ai-analytic-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加redis锁

Browse Source
This commit is contained in:
zren25
2025-04-28 22:01:14 +08:00
parent 96dec58d11
commit bfe30983e3
2 changed files with 72 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
ai-analytic-center-biz/pom.xml
View File

@@ -244,6 +244,18 @@
<version>1.9</version> <!-- 请根据需要选择合适的版本 -->
</dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
<version>3.16.4</version>
<exclusions>
<exclusion>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-redis</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<build>

93
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/AnalysisDifyMqConsumer.java
View File

@@ -14,6 +14,8 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
@@ -23,6 +25,7 @@ import org.springframework.web.bind.annotation.RestController;
import java.time.Instant;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
/**
* @ClassName AnalysisDifyMqConsumer
@@ -38,7 +41,7 @@ import java.util.concurrent.CompletableFuture;
@RestController
@RocketMQMessageListener(topic = "${rocketmq.consumer.analysisDify.topic}",consumerGroup = "${rocketmq.consumer.analysisDify.group}",
instanceName = "analysisDifyMqConsumer",
consumeThreadNumber = 1,
consumeThreadNumber = 20,
enableMsgTrace = true)
public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt> {
@@ -68,44 +71,68 @@ public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt> {
@Autowired
private RedisZSetUtil redisZSetUtil;
@Autowired
private RedissonClient redissonClient;
@Override
public void onMessage(MessageExt messageExt) {
log.info("analysisDifyMqConsumer 当前线程: {}, 线程ID: {}", Thread.currentThread().getName(), Thread.currentThread().getId());
long startTime = System.currentTimeMillis();
Long count = redisZSetUtil.zCard(ConstantStr.DIFY_COUNTERRATELIMIT);
log.info("redis计数数量: {}", count);
if(count>=difyLimit){
log.info("analysisDifyMqConsumer 请求dify超过基数 ：{}，稍后请求: " + redisZSetUtil.getValidMembers(ConstantStr.DIFY_COUNTERRATELIMIT));
throw new RuntimeException("请求dify超过基数 " );
}
String message = new String(messageExt.getBody());
log.info("analysisDifyMqConsumer message: " + message);
DiFyReq difyReq = JSONObject.parseObject(message, DiFyReq.class);
JSONObject difyRequest = JSONObject.parseObject(JSONObject.toJSONString(difyReq.getInputs()), JSONObject.class);
String aiAnalysisRequestId = difyRequest.getString("aiAnalysisRequestId");
long expireTime = Instant.now().plusSeconds(expire).toEpochMilli(); // 60秒后过期
redisZSetUtil.addWithExpire(ConstantStr.DIFY_COUNTERRATELIMIT, aiAnalysisRequestId, expireTime);
RLock lock = redissonClient.getLock("LOCK_PREFIX:" + ConstantStr.DIFY_COUNTERRATELIMIT);
try {
CompletableFuture<JSONObject> future = diFyService.asyncExecuteDifyFlow(difyReq);
future.thenAccept(result -> {
log.info("异步处理asyncExecuteDifyFlow完成aiAnalysisRequestId: {},处理结果：{}", aiAnalysisRequestId, result);
JSONObject data = result.getJSONObject("data");
if(null == result || data.get("status").equals("failed")){
aiAnalysisErrorsService.saveAiAnalysisErrors(AiAnalysisErrors.builder()
.aiAnalysisRequestId(aiAnalysisRequestId)
.aiAnalysisRequestType(difyRequest.getString("aiAnalysisRequestType"))
.aiAnalysisErrorHandlingStatus("0")
.aiAnalysisErrorMessage(data.getString("error"))
.build());
// 2. 尝试加锁（参数说明：等待时间, 锁自动释放时间, 时间单位）
boolean isLocked = lock.tryLock(10, expire, TimeUnit.SECONDS); // 不等待，立即尝试获取
if (isLocked) {
Long count = redisZSetUtil.zCard(ConstantStr.DIFY_COUNTERRATELIMIT);
log.info("redis计数数量: {}", count);
if(count>=difyLimit){
log.info("analysisDifyMqConsumer 请求dify超过基数 ：{}，稍后请求: " + redisZSetUtil.getValidMembers(ConstantStr.DIFY_COUNTERRATELIMIT));
throw new RuntimeException("请求dify超过基数 " );
}
}).exceptionally(ex -> {
log.error("异步处理asyncExecuteDifyFlow 失败aiAnalysisRequestId: {} ,{}", aiAnalysisRequestId,ex.getMessage());
return null;
});
log.info("analysisDifyMqConsumer处理完成，耗时：{}", System.currentTimeMillis() - startTime);
} catch (Exception e) {
log.info(" analysisDifyMqConsumer 处理失败：{}", e.getMessage());
String message = new String(messageExt.getBody());
log.info("analysisDifyMqConsumer message: " + message);
DiFyReq difyReq = JSONObject.parseObject(message, DiFyReq.class);
JSONObject difyRequest = JSONObject.parseObject(JSONObject.toJSONString(difyReq.getInputs()), JSONObject.class);
String aiAnalysisRequestId = difyRequest.getString("aiAnalysisRequestId");
long expireTime = Instant.now().plusSeconds(expire).toEpochMilli(); // 60秒后过期
redisZSetUtil.addWithExpire(ConstantStr.DIFY_COUNTERRATELIMIT, aiAnalysisRequestId, expireTime);
CompletableFuture<JSONObject> future = diFyService.asyncExecuteDifyFlow(difyReq);
future.thenAccept(result -> {
log.info("异步处理asyncExecuteDifyFlow完成aiAnalysisRequestId: {},处理结果：{}", aiAnalysisRequestId, result);
JSONObject data = result.getJSONObject("data");
if(null == result || data.get("status").equals("failed")){
aiAnalysisErrorsService.saveAiAnalysisErrors(AiAnalysisErrors.builder()
.aiAnalysisRequestId(aiAnalysisRequestId)
.aiAnalysisRequestType(difyRequest.getString("aiAnalysisRequestType"))
.aiAnalysisErrorHandlingStatus("0")
.aiAnalysisErrorMessage(data.getString("error"))
.build());
}
}).exceptionally(ex -> {
log.error("异步处理asyncExecuteDifyFlow 失败aiAnalysisRequestId: {} ,{}", aiAnalysisRequestId,ex.getMessage());
return null;
});
lock.unlock();
log.info("analysisDifyMqConsumer处理完成，耗时：{}", System.currentTimeMillis() - startTime);
}else{
log.info("redis锁未获取到 ");
throw new RuntimeException("锁超时 " );
}
} catch (InterruptedException e) {
// Thread.currentThread().interrupt(); // 保持中断状态
throw new RuntimeException("锁获取被中断", e);
} catch (Exception e){
log.error("analysisDifyMqConsumer 异常：{}", e.getMessage());
throw new RuntimeException("请求dify超过基数 " );
}finally {
// 4. 确保只有持有锁的线程才能解锁（关键！）
if (lock != null && lock.isHeldByCurrentThread()) {
lock.unlock();
// 建议记录解锁日志
log.info("释放锁成功，锁KEY: {}", ConstantStr.DIFY_COUNTERRATELIMIT);
}
}
}