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

修改计数

Browse Source
This commit is contained in:
zren25
2025-04-24 19:23:14 +08:00
parent 6c716701ad
commit 4cde67da33
2 changed files with 26 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,6 +17,8 @@ import org.apache.rocketmq.spring.core.RocketMQListener;
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.cloud.context.config.annotation.RefreshScope; import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RestController; import org.springframework.web.bind.annotation.RestController;
@@ -63,23 +65,30 @@ public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt> {
private final ObjectMapper objectMapper = new ObjectMapper(); private final ObjectMapper objectMapper = new ObjectMapper();
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Override @Override
public void onMessage(MessageExt messageExt) { public void onMessage(MessageExt messageExt) {
log.info("analysisDifyMqConsumer 当前线程: {}, 线程ID: {}", Thread.currentThread().getName(), Thread.currentThread().getId()); log.info("analysisDifyMqConsumer 当前线程: {}, 线程ID: {}", Thread.currentThread().getName(), Thread.currentThread().getId());
long startTime = System.currentTimeMillis(); long startTime = System.currentTimeMillis();
log.info("redis计数数量: " + redisCounterRateLimiter.getCurrentCount(ConstantStr.DIFY_COUNTERRATELIMIT)); ZSetOperations zSetOperations = redisTemplate.opsForZSet();
boolean flag = redisCounterRateLimiter.incrementAndCheck(ConstantStr.DIFY_COUNTERRATELIMIT, difyLimit, expire); Long count = zSetOperations.zCard(ConstantStr.DIFY_COUNTERRATELIMIT);
if(!flag){ log.info("redis计数数量: " + count);
if(count>=difyLimit){
log.info("analysisDifyMqConsumer 请求dify超过基数 ：{}，稍后请求: " + redisCounterRateLimiter.getCurrentCount(ConstantStr.DIFY_COUNTERRATELIMIT)); log.info("analysisDifyMqConsumer 请求dify超过基数 ：{}，稍后请求: " + redisCounterRateLimiter.getCurrentCount(ConstantStr.DIFY_COUNTERRATELIMIT));
throw new RuntimeException("请求dify超过基数 " ); 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");
zSetOperations.add(ConstantStr.DIFY_COUNTERRATELIMIT, aiAnalysisRequestId, startTime);
zSetOperations.removeRangeByScore(ConstantStr.DIFY_COUNTERRATELIMIT, 0, startTime - (expire * 1000));
try { try {
String message = new String(messageExt.getBody()); CompletableFuture<JSONObject> future = diFyService.asyncExecuteDifyFlow(difyReq);
log.info("analysisDifyMqConsumer message: " + message);
DiFyReq difyReq = JSONObject.parseObject(message, DiFyReq.class);
CompletableFuture<JSONObject> future = diFyService.asyncExecuteDifyFlow(difyReq);
JSONObject difyRequest = JSONObject.parseObject(JSONObject.toJSONString(difyReq.getInputs()), JSONObject.class);
String aiAnalysisRequestId = difyRequest.getString("aiAnalysisRequestId");
// JSONObject json = future.get(); // JSONObject json = future.get();
future.thenAccept(result -> { future.thenAccept(result -> {
log.info("异步处理asyncExecuteDifyFlow完成aiAnalysisRequestId: {},处理结果：{}", aiAnalysisRequestId, result); log.info("异步处理asyncExecuteDifyFlow完成aiAnalysisRequestId: {},处理结果：{}", aiAnalysisRequestId, result);

11
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/AiAnalysisDifyServiceImpl.java
View File

@@ -25,6 +25,7 @@ import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.apache.rocketmq.spring.core.RocketMQTemplate;
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.data.redis.core.RedisTemplate;
import org.springframework.messaging.support.MessageBuilder; import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@@ -60,14 +61,18 @@ public class AiAnalysisDifyServiceImpl implements AiAnalysisDifyService {
@Autowired @Autowired
private RedisCounterRateLimiter redisCounterRateLimiter; private RedisCounterRateLimiter redisCounterRateLimiter;
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Override @Override
public boolean updateAiDifyResult(String message) { public boolean updateAiDifyResult(String message) {
// 计数-1
redisCounterRateLimiter.decrement(ConstantStr.DIFY_COUNTERRATELIMIT);
log.info("redisdecrement计数数量: " + redisCounterRateLimiter.getCurrentCount(ConstantStr.DIFY_COUNTERRATELIMIT));
if(StringUtils.isNotEmpty(message)){ if(StringUtils.isNotEmpty(message)){
AnalysisDifyResultDTO analysisResp = JSONObject.parseObject(message, AnalysisDifyResultDTO.class); AnalysisDifyResultDTO analysisResp = JSONObject.parseObject(message, AnalysisDifyResultDTO.class);
// 计数-1
Long removedCount = redisTemplate.opsForZSet().remove(ConstantStr.DIFY_COUNTERRATELIMIT, analysisResp.getAiAnalysisRequestId());
log.info("redisdecrement计数数量: " + removedCount);
AiAnalysisRequestLogs oldAiAnalysisRequestLogs = Optional.ofNullable(aiAnalysisRequestLogsService.queryByAiAnalysisRequestId(analysisResp.getAiAnalysisRequestId())) AiAnalysisRequestLogs oldAiAnalysisRequestLogs = Optional.ofNullable(aiAnalysisRequestLogsService.queryByAiAnalysisRequestId(analysisResp.getAiAnalysisRequestId()))
.orElseThrow(() -> new IllegalArgumentException("AiAnalysisRequestId查询对象为空！")); .orElseThrow(() -> new IllegalArgumentException("AiAnalysisRequestId查询对象为空！"));