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

Merge remote-tracking branch 'origin/dev_20250409_difyResult' into feature_20250521_nameplate_difyResult

Browse Source 
# Conflicts:
#	ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/ConstantStr.java
This commit is contained in:
zren25
2025-05-21 11:03:09 +08:00
parent 5453d1a9f2 18b9fdac69
commit adccdec5a2
24 changed files with 1153 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/req/AnalysisQueryReq.java Normal file
View File

@@ -0,0 +1,27 @@
package com.volvo.ai.analytic.center.dto.req;
import lombok.Data;
import javax.validation.constraints.NotNull;
/**
*
* @ClassName: AnalysisRequest
* @author: renzhen
* @Description: Ai查询请求参数
* @date: 2025-04-15 13:39
*/
@Data
public class AnalysisQueryReq {
private Object data;
// aiId
@NotNull(message = "aiAnalysisRequestId不能为空")
private String aiAnalysisRequestId;
// 是否重试：默认：否
private String retryAnalyze;
}

56
ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/req/AnalysisReq.java Normal file
View File

@@ -0,0 +1,56 @@
package com.volvo.ai.analytic.center.dto.req;
import com.alibaba.csp.sentinel.util.StringUtil;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import javax.validation.Valid;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
/**
*
* @ClassName: AnalysisRequest
* @author: renzhen
* @Description: Ai解析请求参数
* @date: 2025-04-15 13:39
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class AnalysisReq {
// 请求的所有参数
@Valid
@NotNull(message = "请求语料不能为空")
private Object data;
// 回调地址
@Valid
@NotBlank(message = "回调地址不能为空")
private String callbackUrl;
// aiId
private String aiAnalysisRequestId;
// 业务类型
@Valid
@NotBlank(message = "业务类型不能为空")
private String aiAnalysisRequestType;
public void validate() {
if (null == data){
throw new IllegalArgumentException("请求语料不能为空");
}
if (StringUtil.isBlank(aiAnalysisRequestType)){
throw new IllegalArgumentException("业务类型不能为空");
}
if (StringUtil.isBlank(callbackUrl)){
throw new IllegalArgumentException("回调地址不能为空");
}
}
}

24
ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/resp/AnalysisDifyResultDTO.java Normal file
View File

@@ -0,0 +1,24 @@
package com.volvo.ai.analytic.center.dto.resp;
import lombok.Data;
/**
*
* @ClassName: AnalysisRequest
* @author: renzhen
* @Description: 工作流处理结果
* @date: 2025-04-15 13:39
*/
@Data
public class AnalysisDifyResultDTO{
// 响应
private String workflowRunId;
private String workflowAppId;
private String workUserId;
// 分析中心唯一ID
private String aiAnalysisRequestId;
private String difyResponse;
private String aiAnalysisRequestType;
}

57
ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/resp/AnalysisResp.java Normal file
View File

@@ -0,0 +1,57 @@
package com.volvo.ai.analytic.center.dto.resp;
import com.volvo.common.core.constant.CommonConstants;
import com.volvo.common.core.util.ResultMsg;
import lombok.Data;
/**
*
* @ClassName: AnalysisRequest
* @author: renzhen
* @Description: Ai解析响应
* @date: 2025-04-15 13:39
*/
@Data
public class AnalysisResp<T> {
// 响应
private T data;
// 分析中心唯一ID
private String aiAnalysisRequestId;
private int code;
private String msg;
public static <T> AnalysisResp<T> success(String message) {
return (AnalysisResp<T>) analysisResp((Object)null, CommonConstants.SUCCESS, message);
}
public static <T> AnalysisResp<T> success(T data, String aiAnalysisRequestId) {
AnalysisResp<T> analysisResp = new AnalysisResp();
analysisResp.setAiAnalysisRequestId(aiAnalysisRequestId);
analysisResp.setData(data);
analysisResp.setMsg("ok");
analysisResp.setCode(CommonConstants.SUCCESS);
return analysisResp;
}
public static <T> AnalysisResp<T> failed(String message) {
return (AnalysisResp<T>) analysisResp((Object)null, CommonConstants.FAIL, message);
}
private static <T> AnalysisResp<T> analysisResp(T data, int code, String msg) {
AnalysisResp<T> analysisResp = new AnalysisResp();
analysisResp.setCode(code);
analysisResp.setData(data);
analysisResp.setMsg(msg);
return analysisResp;
}
private static <T> AnalysisResp<T> analysisResp(T data, String aiAnalysisRequestId) {
AnalysisResp<T> analysisResp = new AnalysisResp();
analysisResp.setAiAnalysisRequestId(aiAnalysisRequestId);
analysisResp.setData(data);
analysisResp.setMsg("ok");
analysisResp.setCode(CommonConstants.SUCCESS);
return analysisResp;
}
}

12
ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/entity/AiAnalysisRequestLogs.java
View File

@@ -43,6 +43,18 @@ public class AiAnalysisRequestLogs {
@TableField("dify_agent_key")
private String difyAgentKey;
@TableField("workflow_run_id")
private String workflowRunId;
@TableField("workflow_app_id")
private String workflowAppId;
@TableField("work_user_id")
private String workUserId;
@TableField("callback_url")
private String callbackUrl;
@TableField("is_deleted")
@TableLogic
private Integer isDeleted;

70
ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/entity/TcBusinessType.java Normal file
View File

@@ -0,0 +1,70 @@
package com.volvo.ai.analytic.center.entity;
import com.baomidou.mybatisplus.annotation.*;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.Date;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@TableName("tc_business_type")
public class TcBusinessType {
@TableId(value = "id", type = IdType.AUTO)
private Long id;
@TableField("business_request_type")
private String businessRequestType;
@TableField("business_request_desc")
private String businessRequestDesc;
@TableField("business_type_topic")
private String businessTypeTopic; // JSON 字符串
@TableField("business_type_topic_tag")
private String businessTypeTopicTag;
@TableField("workflow_api_key")
private String workflowApiKey; // JSON 字符串
@TableField("workflow_user")
private String workflowUser;
@TableField("is_deleted")
@TableLogic
private Integer isDeleted;
@TableField("versions")
@Version
private Integer versions;
/**
* 创建者
*/
@TableField("create_by")
private String createBy;
/**
* 创建时间
*/
@TableField("create_time")
private Date createTime;
/**
* 更新者
*/
@TableField("update_by")
private String updateBy;
/**
* 更新时间
*/
@TableField("update_time")
private Date updateTime;
}

2
ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/feign/DiFyFeign.java
View File

@@ -21,7 +21,7 @@ public interface DiFyFeign {
@PostMapping(value = "/v1/files/upload" , consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
JSONObject fileUpload(@RequestHeader(value = "Authorization") String authorization, @RequestPart("file") MultipartFile file);
@PostMapping(value = "/v1/workflows/run/:{workflowRunId}" , consumes = MediaType.APPLICATION_JSON_VALUE
@GetMapping(value = "/v1/workflows/run/{workflowRunId}" , consumes = MediaType.APPLICATION_JSON_VALUE
, produces = MediaType.APPLICATION_JSON_VALUE)
JSONObject queryWorkFlowById(@RequestHeader(value = "Authorization") String authorization, @PathVariable("workflowRunId") String workflowRunId);
}

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

@@ -232,6 +232,11 @@
<version>2.3.0</version>
</dependency>
<!-- Redis Starter -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
@@ -239,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>

81
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/AiAnalysisDifyController.java Normal file
View File

@@ -0,0 +1,81 @@
package com.volvo.ai.analytic.center.controller;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.volvo.ai.analytic.center.dto.req.AnalysisQueryReq;
import com.volvo.ai.analytic.center.dto.req.AnalysisReq;
import com.volvo.ai.analytic.center.dto.resp.AnalysisResp;
import com.volvo.ai.analytic.center.service.AiAnalysisDifyService;
import com.volvo.common.core.util.ResultMsg;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController()
@Api(tags = "分析中心接口")
@Slf4j
@RefreshScope
@RequestMapping("analysis")
public class AiAnalysisDifyController {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Autowired
private AiAnalysisDifyService aiAnalysisDifyService;
@PostMapping("/updateByAiId")
@ApiOperation(value = "更新dify结果")
public ResultMsg<Object> updateByAiId(@RequestBody String message) {
log.info("updateByAiId message: {}", message);
aiAnalysisDifyService.updateAiDifyResult(message);
return ResultMsg.ok("ok");
}
@PostMapping("/aiAnalyze")
@ApiOperation(value = "Ai解析接口")
public AnalysisResp<Object> aiAnalyze(@Validated @RequestBody String message) {
log.info("aiAnalyze data: {}",message);
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true); // 允许未转义的控制字符
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
try {
AnalysisReq analysisReq = objectMapper.readValue(message, AnalysisReq.class);
analysisReq.validate();
return aiAnalysisDifyService.aiAnalyze(analysisReq);
} catch (JsonProcessingException e) {
log.info("aiAnalyze data error: {}",e);
return AnalysisResp.failed("解析失败");
}
}
@PostMapping("/query")
@ApiOperation(value = "Ai解析结果查询")
public AnalysisResp<Object> query(@RequestBody @Validated AnalysisQueryReq analysisQueryReq) {
log.info("aiAnalyze query data: {}",analysisQueryReq);
return aiAnalysisDifyService.query(analysisQueryReq);
}
@PostMapping("/callback")
@ApiOperation(value = "Ai解析结果查询")
public AnalysisResp<Object> testCallback(@RequestBody AnalysisResp analysisResp) {
log.info("aiAnalyze testCallback data: {}",analysisResp);
return AnalysisResp.success("ok");
}
}

48
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/AiDifyResultController.java
View File

@@ -1,48 +0,0 @@
package com.volvo.ai.analytic.center.controller;
import com.alibaba.fastjson.JSONObject;
import com.volvo.ai.analytic.center.entity.AiAnalysisRequestLogs;
import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
import com.volvo.common.core.util.ResultMsg;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@Api(tags = "AiDifyResult")
@RequestMapping("")
@Slf4j
@RefreshScope
public class AiDifyResultController {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Autowired
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
@PostMapping("/updateByAiId")
@ApiOperation(value = "更新dify结果")
public ResultMsg<Object> updateByAiId(@RequestBody String message) {
JSONObject messageJson = JSONObject.parseObject(message);
AiAnalysisRequestLogs aiAnalysisRequestLogs = new AiAnalysisRequestLogs();
aiAnalysisRequestLogs.setAiAnalysisRequestId(messageJson.getString("aiAnalysisRequestId"));
aiAnalysisRequestLogs.setDifyResponse(messageJson.getString("difyResponse"));
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(aiAnalysisRequestLogs);
return ResultMsg.ok("ok");
}
}

80
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/RedisTestController.java Normal file
View File

@@ -0,0 +1,80 @@
package com.volvo.ai.analytic.center.controller;
import com.alibaba.fastjson.JSONObject;
import com.volvo.ai.analytic.center.service.AiAnalysisDifyService;
import com.volvo.ai.analytic.center.utils.RedisZSetUtil;
import com.volvo.common.core.util.ResultMsg;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.Instant;
import java.util.Set;
@RestController()
@Api(tags = "分析中心接口")
@Slf4j
@RefreshScope
@RequestMapping("redis")
public class RedisTestController {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Autowired
private AiAnalysisDifyService aiAnalysisDifyService;
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private RedisZSetUtil redisZSetUtil;
@PostMapping("/testRedis")
public void test(@RequestBody String message) {
String zsetKey = "myZSet";
JSONObject json = JSONObject.parseObject(message);
// 添加成员并设置过期时间戳
long expireTime1 = Instant.now().plusSeconds(30).toEpochMilli(); // 30秒后过期
long expireTime2 = Instant.now().plusSeconds(60).toEpochMilli(); // 60秒后过期
redisZSetUtil.addWithExpire(zsetKey, "member1", expireTime1);
redisZSetUtil.addWithExpire(zsetKey, "member2", expireTime2);
// 获取未过期的成员
Set<String> validMembers = redisZSetUtil.getValidMembers(zsetKey);
log.info("Valid members: " + validMembers);
}
@PostMapping("/queryRedis")
@ApiOperation(value = "queryRedis")
public ResultMsg<Object> queryRedis(@RequestBody String key) {
log.info("aiAnalyze queryRedis : {}", redisTemplate.opsForZSet().zCard(key));
Set<String> getredisSet = redisTemplate.opsForZSet().range(key, 0, -1);
log.info("queryRedis:{}", getredisSet);
return ResultMsg.ok(getredisSet);
}
@PostMapping("/del")
@ApiOperation(value = "del")
public ResultMsg<Object> del(@RequestBody String key) {
log.info("aiAnalyze queryRedis : {}",key);
redisTemplate.opsForZSet().removeRange(key, 0, -1);
Set<String> getRedisSet = redisTemplate.opsForZSet().range(key, 0, -1);
log.info("removeRange:{}" , getRedisSet);
redisTemplate.delete(key);
Set<String> getredisSet2 = redisTemplate.opsForZSet().range(key, 0, -1);
log.info("delete:{}", getredisSet2);
return ResultMsg.ok(getredisSet2);
}
}

74
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/AiAnalysisDifyJob.java Normal file
View File

@@ -0,0 +1,74 @@
package com.volvo.ai.analytic.center.job;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.volvo.ai.analytic.center.entity.AiAnalysisRequestLogs;
import com.volvo.ai.analytic.center.feign.DiFyFeign;
import com.volvo.ai.analytic.center.mapper.AiAnalysisRequestLogsMapper;
import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
import com.volvo.common.core.util.ResultMsg;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import java.util.Arrays;
import java.util.List;
/**
*
*/
@Slf4j
@Component
@RestController
public class AiAnalysisDifyJob {
@Autowired
private AiAnalysisRequestLogsMapper aiAnalysisRequestLogsMapper;
@Autowired
private DiFyFeign diFyFeign;
@Autowired
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
/**
* 失败的查询处理
*/
@XxlJob("workflowRunIdFaile")
@PostMapping("workflowRunIdFaile")
public ResultMsg workflowRunIdFaile(@RequestBody String paramJson) {
try {
// 获取任务参数
String param = XxlJobHelper.getJobParam();
if(StringUtils.isEmpty(param)){
param = paramJson;
}
List<String> workflowRunIdList = Arrays.asList(param.split(","));
LambdaQueryWrapper<AiAnalysisRequestLogs> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.in(AiAnalysisRequestLogs::getWorkflowRunId, workflowRunIdList);
queryWrapper.eq(AiAnalysisRequestLogs::getIsDeleted, "0");
List<AiAnalysisRequestLogs> aiAnalysisRequestLogsList = aiAnalysisRequestLogsMapper.selectList(queryWrapper);
aiAnalysisRequestLogsList.stream().forEach(aiAnalysisRequestLogs -> {
JSONObject jsonResult = diFyFeign.queryWorkFlowById("Bearer "+aiAnalysisRequestLogs.getDifyAgentKey(),aiAnalysisRequestLogs.getWorkflowRunId());
String outputs = jsonResult.getString("outputs");
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder()
.aiAnalysisRequestId(aiAnalysisRequestLogs.getAiAnalysisRequestId())
.difyResponse(outputs)
.build());
});
} catch (Exception e) {
log.error("processMessageByTask 定时任务补偿处理消息异常",e.getMessage());
throw new RuntimeException(e);
}
return ResultMsg.ok();
}
}

56
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/DccCorpusJob.java
View File

@@ -1,56 +0,0 @@
package com.volvo.ai.analytic.center.job;
import com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper;
import com.volvo.ai.analytic.center.service.TmOdsVdqwMessagearchivingService;
import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
import com.volvo.common.core.util.ResultMsg;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
@Slf4j
@Component
@RestController
public class DccCorpusJob {
@Autowired
private TmOdsVdqwMessagearchivingService tmOdsVdqwMessagearchivingService;
@Autowired
private TmTelephoneCorpusService tmTelephoneCorpusService;
@Autowired
private TmTelephoneCorpusMapper tmTelephoneCorpusMapper;
/**
* dcc语料处理
*/
@XxlJob("dccCorpusJob")
public ResultMsg dccCorpusJob(@RequestBody String paramJson) {
try {
// 获取任务参数
String param = XxlJobHelper.getJobParam();
if(StringUtils.isEmpty(param)){
param = paramJson;
}
// 分页查询 过滤已跑批并发送的
tmOdsVdqwMessagearchivingService.runQiWeiCorpusDify(param);
} catch (Exception e) {
log.error("processMessageByTask 定时任务补偿处理消息异常",e.getMessage());
throw new RuntimeException(e);
}
return ResultMsg.ok();
}
}

11
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mapper/TcBusinessTypeMapper.java Normal file
View File

@@ -0,0 +1,11 @@
package com.volvo.ai.analytic.center.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.volvo.ai.analytic.center.entity.TcBusinessType;
import org.apache.ibatis.annotations.Mapper;
@Mapper
public interface TcBusinessTypeMapper extends BaseMapper<TcBusinessType> {
}

91
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/AnalysisDifyCallbackMqConsumer.java Normal file
View File

@@ -0,0 +1,91 @@
package com.volvo.ai.analytic.center.mq;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.volvo.ai.analytic.center.dto.resp.AnalysisDifyResultDTO;
import com.volvo.ai.analytic.center.dto.resp.AnalysisResp;
import com.volvo.ai.analytic.center.entity.AiAnalysisRequestLogs;
import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
import com.volvo.ai.analytic.center.service.DiFyService;
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.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;
/**
* @ClassName AnalysisDifyMqConsumer
* @Description AI解析MQ-Callback处理
* @Author renzhen
* @Date 2025-03-04 10:18
* @Version 1.0
**/
@Slf4j
@Component
@RefreshScope
@RestController
@RocketMQMessageListener(topic = "${rocketmq.consumer.analysisDify.callbackTopic}",consumerGroup = "${rocketmq.consumer.analysisDify.callbackGroup}",
instanceName = "analysisDifyCallbackMqConsumer",
consumeThreadNumber = 40,
enableMsgTrace = true)
public class AnalysisDifyCallbackMqConsumer implements RocketMQListener<MessageExt> {
@Autowired
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
@Value("${dify.corpus.checkDccRepeat}")
private String checkDccRepeat;
@Autowired
private DiFyService diFyService;
@Autowired
private RestTemplate restTemplate;
private final ObjectMapper objectMapper = new ObjectMapper();
@Override
public void onMessage(MessageExt messageExt) {
long startTime = System.currentTimeMillis();
try {
log.info("analysisDifyCallbackMqConsumer 当前线程: {}, 线程ID: {}", Thread.currentThread().getName(), Thread.currentThread().getId());
String message = new String(messageExt.getBody());
log.info("analysisDifyCallbackMqConsumer message: " + message);
AnalysisDifyResultDTO analysisRestDto = JSONObject.parseObject(message, AnalysisDifyResultDTO.class);
AiAnalysisRequestLogs aiAnalysisRequestLogs = aiAnalysisRequestLogsService.queryByAiAnalysisRequestId(analysisRestDto.getAiAnalysisRequestId());
AnalysisResp analysisResp = new AnalysisResp();
analysisResp.setAiAnalysisRequestId(analysisRestDto.getAiAnalysisRequestId());
JSONObject difyJson = JSONObject.parseObject(analysisRestDto.getDifyResponse());
String outputs = difyJson.getString("outputs");
analysisResp.setData(outputs);
HttpHeaders headers = new HttpHeaders();
headers.set("Content-Type", "application/json");
// 封装请求体和请求头
HttpEntity<AnalysisResp> requestEntity = new HttpEntity<>(analysisResp, headers);
ResponseEntity<String> response = restTemplate.postForEntity(aiAnalysisRequestLogs.getCallbackUrl(), requestEntity, String.class); // 响应类型);
if (response.getStatusCode().is2xxSuccessful()) {
log.info("analysisDifyCallbackMqConsumer aiAnalysisRequestId：{},回调请求成功，url:{}: " ,analysisResp.getAiAnalysisRequestId(), aiAnalysisRequestLogs.getCallbackUrl());
} else {
log.info("analysisDifyCallbackMqConsumer aiAnalysisRequestId：{},回调请求失败，url:{}: " ,analysisResp.getAiAnalysisRequestId(), aiAnalysisRequestLogs.getCallbackUrl());
}
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(analysisResp.getAiAnalysisRequestId()).businessResponse(outputs).build());
log.info(" analysisDifyCallbackMqConsumer，耗时：{}", System.currentTimeMillis() - startTime);
} catch (Exception e) {
log.info(" analysisDifyCallbackMqConsumer mq 处理失败：{}", e.getMessage());
}
}
}

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

@@ -0,0 +1,141 @@
package com.volvo.ai.analytic.center.mq;
import com.alibaba.fastjson.JSONObject;
import com.volvo.ai.analytic.center.dto.req.DiFyReq;
import com.volvo.ai.analytic.center.entity.AiAnalysisErrors;
import com.volvo.ai.analytic.center.service.AiAnalysisErrorsService;
import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
import com.volvo.ai.analytic.center.service.DiFyService;
import com.volvo.ai.analytic.center.utils.ConstantStr;
import com.volvo.ai.analytic.center.utils.RedisCounterRateLimiter;
import com.volvo.ai.analytic.center.utils.RedisLockService;
import com.volvo.ai.analytic.center.utils.RedisZSetUtil;
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.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RestController;
import java.time.Instant;
import java.util.concurrent.CompletableFuture;
/**
* @ClassName AnalysisDifyMqConsumer
* @Description AI解析 MQ处理
* @Author renzhen
* @Date 2025-03-04 10:18
* @Version 1.0
**/
@Slf4j
@Component
@RefreshScope
@RestController
@RocketMQMessageListener(topic = "${rocketmq.consumer.analysisDify.topic}",consumerGroup = "${rocketmq.consumer.analysisDify.group}",
instanceName = "analysisDifyMqConsumer",
consumeThreadNumber = 20,
enableMsgTrace = true)
public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt> {
@Autowired
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
@Value("${dify.corpus.checkDccRepeat}")
private String checkDccRepeat;
@Value("${rocketmq.consumer.analysisDify.difyLimit}")
private int difyLimit;
@Value("${rocketmq.consumer.analysisDify.expire}")
private Long expire;
@Autowired
private DiFyService diFyService;
@Autowired
private RedisCounterRateLimiter redisCounterRateLimiter;
@Autowired
private AiAnalysisErrorsService aiAnalysisErrorsService;
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private RedisZSetUtil redisZSetUtil;
@Autowired
private RedisLockService redisLockService;
@Override
public void onMessage(MessageExt messageExt) {
log.info("analysisDifyMqConsumer 当前线程: {}, 线程ID: {}", Thread.currentThread().getName(), Thread.currentThread().getId());
long startTime = System.currentTimeMillis();
String lockKey = "LOCK_PREFIX:" + ConstantStr.DIFY_COUNTERRATELIMIT;
// RLock lock = redissonClient.getLock("LOCK_PREFIX:" + ConstantStr.DIFY_COUNTERRATELIMIT);
try {
boolean isLocked = redisLockService.tryLock(lockKey,ConstantStr.DIFY_COUNTERRATELIMIT, expire); // 不等待，立即尝试获取
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超过基数 " );
}
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 -> {
aiAnalysisErrorsService.saveAiAnalysisErrors(AiAnalysisErrors.builder()
.aiAnalysisRequestId(aiAnalysisRequestId)
.aiAnalysisRequestType(difyRequest.getString("aiAnalysisRequestType"))
.aiAnalysisErrorHandlingStatus("0")
.aiAnalysisErrorMessage(ex.getMessage())
.build());
log.error("异步处理asyncExecuteDifyFlow 失败aiAnalysisRequestId: {} ,{}", aiAnalysisRequestId,ex.getMessage());
return null;
});
redisLockService.releaseLock(lockKey, ConstantStr.DIFY_COUNTERRATELIMIT);
log.info("analysisDifyMqConsumer处理完成，耗时：{}", System.currentTimeMillis() - startTime);
}else{
log.info("redis锁未获取到 ");
throw new RuntimeException("锁超时 " );
}
} catch (Exception e){
log.error("analysisDifyMqConsumer 异常：{}", e.getMessage());
throw new RuntimeException("请求dify超过基数 " );
}finally {
redisLockService.releaseLock(lockKey, ConstantStr.DIFY_COUNTERRATELIMIT);
// 建议记录解锁日志
log.info("释放锁成功，锁KEY: {}", ConstantStr.DIFY_COUNTERRATELIMIT);
}
}
}

19
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/AiAnalysisDifyService.java Normal file
View File

@@ -0,0 +1,19 @@
package com.volvo.ai.analytic.center.service;
import com.volvo.ai.analytic.center.dto.req.AnalysisQueryReq;
import com.volvo.ai.analytic.center.dto.req.AnalysisReq;
import com.volvo.ai.analytic.center.dto.resp.AnalysisDifyResultDTO;
import com.volvo.ai.analytic.center.dto.resp.AnalysisResp;
public interface AiAnalysisDifyService {
public boolean updateAiDifyResult(String message);
AnalysisResp aiAnalyze(AnalysisReq analysisReq);
AnalysisResp query(AnalysisQueryReq analysisQueryReq);
}

4
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/DiFyService.java
View File

@@ -3,6 +3,8 @@ package com.volvo.ai.analytic.center.service;
import com.alibaba.fastjson.JSONObject;
import com.volvo.ai.analytic.center.dto.req.DiFyReq;
import java.util.concurrent.CompletableFuture;
public interface DiFyService {
@@ -11,4 +13,6 @@ public interface DiFyService {
public JSONObject executeDifyFlow(DiFyReq diFyReq, String businessType, String businessData, String aiAnalysisRequestId);
public JSONObject executeDifyFlow(DiFyReq diFyReq);
public CompletableFuture<JSONObject> asyncExecuteDifyFlow(DiFyReq diFyReq);
}

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

@@ -0,0 +1,204 @@
package com.volvo.ai.analytic.center.service.impl;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.volvo.ai.analytic.center.dto.req.AnalysisQueryReq;
import com.volvo.ai.analytic.center.dto.req.AnalysisReq;
import com.volvo.ai.analytic.center.dto.req.DiFyReq;
import com.volvo.ai.analytic.center.dto.resp.AnalysisDifyResultDTO;
import com.volvo.ai.analytic.center.dto.resp.AnalysisResp;
import com.volvo.ai.analytic.center.entity.AiAnalysisRequestLogs;
import com.volvo.ai.analytic.center.entity.TcBusinessType;
import com.volvo.ai.analytic.center.mapper.TcBusinessTypeMapper;
import com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper;
import com.volvo.ai.analytic.center.service.AiAnalysisDifyService;
import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
import com.volvo.ai.analytic.center.utils.AiAnalysisUtils;
import com.volvo.ai.analytic.center.utils.ConstantStr;
import com.volvo.ai.analytic.center.utils.RedisCounterRateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@Slf4j
@Service
public class AiAnalysisDifyServiceImpl implements AiAnalysisDifyService {
@Autowired
private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
@Autowired
private TmTelephoneCorpusMapper tmTelephoneCorpusMapper;
@Autowired
private TmTelephoneCorpusService tmTelephoneCorpusService;
@Autowired
private TcBusinessTypeMapper tcBusinessTypeMapper;
@Value("${rocketmq.producer.analysisDify.topic}")
private String analysisDifyTopic;
@Value("${rocketmq.producer.analysisDify.callbackTopic}")
private String callbackTopic;
@Resource
private RocketMQTemplate rocketMqTemplate;
@Autowired
private RedisCounterRateLimiter redisCounterRateLimiter;
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Override
public boolean updateAiDifyResult(String message) {
if(StringUtils.isNotEmpty(message)){
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()))
.orElseThrow(() -> new IllegalArgumentException("AiAnalysisRequestId查询对象为空！"));
AiAnalysisRequestLogs aiAnalysisRequestLogs = new AiAnalysisRequestLogs();
aiAnalysisRequestLogs.setAiAnalysisRequestId(analysisResp.getAiAnalysisRequestId());
aiAnalysisRequestLogs.setDifyResponse(analysisResp.getDifyResponse());
aiAnalysisRequestLogs.setWorkflowRunId(analysisResp.getWorkflowRunId());
aiAnalysisRequestLogs.setWorkflowAppId(analysisResp.getWorkflowRunId());
aiAnalysisRequestLogs.setWorkUserId(analysisResp.getWorkUserId());
JSONObject difyJson = JSONObject.parseObject(analysisResp.getDifyResponse());
aiAnalysisRequestLogs.setBusinessResponse(difyJson.getString("outputs"));
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(aiAnalysisRequestLogs);
if(StringUtils.isNotEmpty(oldAiAnalysisRequestLogs.getCallbackUrl())){
// 发送 mq
sendMq(callbackTopic, analysisResp);
}
return true;
}
return false;
}
@Override
public AnalysisResp aiAnalyze(AnalysisReq analysisReq) {
Optional.ofNullable(analysisReq).orElseThrow(() -> {
log.info("请求Ai解析对象为空！");
return new IllegalArgumentException("请求Ai解析对象为空！");
});
String aiAnalysisRequestId = StringUtils.isEmpty(analysisReq.getAiAnalysisRequestId())? AiAnalysisUtils.getAiAnalysisRequestId(analysisReq.getAiAnalysisRequestType()):analysisReq.getAiAnalysisRequestId();
Map<String, TcBusinessType> queryTcBusinessType = queryTcBusinessType();
TcBusinessType tcBusinessType = Optional.ofNullable(queryTcBusinessType.get(analysisReq.getAiAnalysisRequestType()))
.filter(businessType -> StringUtils.isNotEmpty(businessType.getWorkflowApiKey()))
.orElseThrow(() -> {
log.info("接入业务类型未配置！");
return new IllegalArgumentException("接入业务类型未配置！");
});
DiFyReq diFyReq = createDiFyReq(analysisReq, tcBusinessType, aiAnalysisRequestId);
saveAiAnalysisRequestLogs(analysisReq, diFyReq, aiAnalysisRequestId);
sendMq(analysisDifyTopic, diFyReq);
return AnalysisResp.success(analysisReq.getData(),aiAnalysisRequestId);
}
private DiFyReq createDiFyReq(AnalysisReq analysisReq, TcBusinessType tcBusinessType, String aiAnalysisRequestId) {
DiFyReq diFyReq = new DiFyReq();
diFyReq.setUser(StringUtils.isEmpty(tcBusinessType.getWorkflowUser()) ? analysisReq.getAiAnalysisRequestType().concat("_USER") : tcBusinessType.getWorkflowUser());
diFyReq.setFlowId(tcBusinessType.getWorkflowApiKey());
JSONObject difyRequest = JSONObject.parseObject(JSONObject.toJSONString(analysisReq.getData()), JSONObject.class);
difyRequest.put("aiAnalysisRequestId", aiAnalysisRequestId);
difyRequest.put("aiAnalysisRequestType", tcBusinessType.getBusinessRequestType() );
diFyReq.setInputs(difyRequest);
return diFyReq;
}
private void saveAiAnalysisRequestLogs(AnalysisReq analysisReq, DiFyReq diFyReq, String aiAnalysisRequestId) {
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder()
.aiAnalysisRequestId(aiAnalysisRequestId)
.businessRequest(JSONObject.toJSONString(analysisReq.getData()))
.difyAgentKey(diFyReq.getFlowId())
.difyRequest(JSON.toJSONString(diFyReq))
.aiAnalysisRequestType(analysisReq.getAiAnalysisRequestType())
.callbackUrl(analysisReq.getCallbackUrl())
.build());
}
@Override
public AnalysisResp query(AnalysisQueryReq analysisQueryReq) {
if(null == analysisQueryReq){
log.info("请求Ai查询对象为空！");
return AnalysisResp.failed("请求Ai查询对象为空！");
}
Optional<AiAnalysisRequestLogs> aiAnalysisRequestLogsOpt = Optional.ofNullable(
aiAnalysisRequestLogsService.queryByAiAnalysisRequestId(analysisQueryReq.getAiAnalysisRequestId())
);
return aiAnalysisRequestLogsOpt.map(logs -> {
if ("true".equals(analysisQueryReq.getRetryAnalyze())) {
log.info("请求Ai查询 需要重新生成AI解析 aiAnalysisRequestId:{}, retryAnalyze：{}", analysisQueryReq.getAiAnalysisRequestId(), analysisQueryReq.getRetryAnalyze());
return aiAnalyze(AnalysisReq.builder()
.data(logs.getBusinessRequest())
.aiAnalysisRequestType(logs.getAiAnalysisRequestType())
.callbackUrl(logs.getCallbackUrl())
.build());
}
return AnalysisResp.success(logs.getBusinessResponse(), logs.getAiAnalysisRequestId());
}).orElseGet(() -> {
log.info("查询的AI解析不存在！aiAnalysisRequestId:{}", analysisQueryReq.getAiAnalysisRequestId());
return AnalysisResp.failed("查询的AI解析不存在！");
});
}
public Map<String, TcBusinessType> queryTcBusinessType() {
LambdaQueryWrapper<TcBusinessType> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(TcBusinessType::getIsDeleted, "0");
List<TcBusinessType> tcBusinessTypeList =tcBusinessTypeMapper.selectList(queryWrapper);
return tcBusinessTypeList.stream()
.collect(Collectors.toMap(
TcBusinessType::getBusinessRequestType,
tcBusinessType -> tcBusinessType
));
}
private void sendMq(String topic, Object message){
rocketMqTemplate.asyncSend(topic, MessageBuilder.withPayload(message).build(),
new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
log.info("请求AI解析发送MQ成功消息体:{}", message);
}
@Override
public void onException(Throwable e) {
log.error("请求AI解析发送MQ异常消息体:{}, 异常：", message, e);
}
}, 10000);
}
}

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

@@ -17,6 +17,7 @@ import org.springframework.stereotype.Service;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
@Slf4j
@Service
@@ -117,4 +118,13 @@ public class DiFyServiceImpl implements DiFyService{
return data;
}
@Override
public CompletableFuture<JSONObject> asyncExecuteDifyFlow(DiFyReq diFyReq) {
Map<String, Object> map = new HashMap<>();
map.put("inputs",diFyReq.getInputs());
map.put("user",diFyReq.getUser());
return CompletableFuture.supplyAsync(() -> diFyFeign.runWorkflows("Bearer "+diFyReq.getFlowId(),map));
}
}

2
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/ConstantStr.java
View File

@@ -21,4 +21,6 @@ public class ConstantStr {
public static final String customerFlowIds = "customerFlowIds";
public static final String aiAnalysisRequestId = "aiAnalysisRequestId";
public static final String DIFY_COUNTERRATELIMIT = "DIFY_COUNTERRATELIMIT";
}

80
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisCounterRateLimiter.java Normal file
View File

@@ -0,0 +1,80 @@
package com.volvo.ai.analytic.center.utils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import java.util.Collections;
@Service
public class RedisCounterRateLimiter {
private final StringRedisTemplate redisTemplate;
@Autowired
public RedisCounterRateLimiter(StringRedisTemplate redisTemplate) {
this.redisTemplate = redisTemplate;
}
/**
* 增加计数并检查是否超过限制
* @param key 限流key
* @param limit 最大限制数
* @param expire 过期时间(秒)
* @return true-允许请求; false-超过限制
*/
public boolean incrementAndCheck(String key, int limit, long expire) {
// 使用Lua脚本保证原子性
String luaScript =
"local current = redis.call('GET', KEYS[1]) or '0'\n" +
"local num = tonumber(current)\n" +
"if num >= tonumber(ARGV[1]) then\n" +
" return 0\n" +
"else\n" +
" redis.call('INCR', KEYS[1])\n" +
" if num == 0 then\n" +
" redis.call('EXPIRE', KEYS[1], ARGV[2])\n" +
" end\n" +
" return 1\n" +
"end";
DefaultRedisScript<Long> script = new DefaultRedisScript<>();
script.setScriptText(luaScript);
script.setResultType(Long.class);
Long result = redisTemplate.execute(script, Collections.singletonList(key),
String.valueOf(limit), String.valueOf(expire));
return result != null && result == 1L;
}
/**
* 减少计数
* @param key 限流key
*/
public void decrement(String key) {
// 使用Lua脚本防止减到负数
String luaScript =
"local current = redis.call('GET', KEYS[1]) or '0'\n" +
"if tonumber(current) > 0 then\n" +
" redis.call('DECR', KEYS[1])\n" +
"end\n" +
"return 1";
DefaultRedisScript<Long> script = new DefaultRedisScript<>();
script.setScriptText(luaScript);
script.setResultType(Long.class);
redisTemplate.execute(script, Collections.singletonList(key));
}
/**
* 获取当前计数
* @param key 限流key
* @return 当前计数值
*/
public int getCurrentCount(String key) {
String count = redisTemplate.opsForValue().get(key);
return count == null ? 0 : Integer.parseInt(count);
}
}

43
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisLockService.java Normal file
View File

@@ -0,0 +1,43 @@
package com.volvo.ai.analytic.center.utils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import java.util.concurrent.TimeUnit;
@Service
public class RedisLockService {
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 尝试获取分布式锁
*
* @param lockKey 锁的键名
* @param requestId 请求标识（用于解锁时验证）
* @param expireTime 锁的过期时间（秒）
* @return 是否获取锁成功
*/
public boolean tryLock(String lockKey, String requestId, long expireTime) {
return stringRedisTemplate.opsForValue()
.setIfAbsent(lockKey, requestId, expireTime, TimeUnit.SECONDS);
}
/**
* 释放分布式锁
*
* @param lockKey 锁的键名
* @param requestId 请求标识（用于验证）
* @return 是否释放锁成功
*/
public boolean releaseLock(String lockKey, String requestId) {
String currentValue = stringRedisTemplate.opsForValue().get(lockKey);
if (currentValue != null && currentValue.equals(requestId)) {
stringRedisTemplate.delete(lockKey);
return true;
}
return false;
}
}

49
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisZSetUtil.java Normal file
View File

@@ -0,0 +1,49 @@
package com.volvo.ai.analytic.center.utils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Instant;
import java.util.Set;
@Service
public class RedisZSetUtil {
// @Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
public RedisZSetUtil(RedisTemplate<String, String> redisTemplate) {
this.redisTemplate = redisTemplate;
}
/**
* 添加成员到 ZSet，并设置过期时间戳
*/
public void addWithExpire(String zsetKey, String member, long expireTimeMillis) {
redisTemplate.opsForZSet().add(zsetKey, member, expireTimeMillis);
}
/**
* 清理过期的成员
*/
@Scheduled(fixedRate = 1000)
public void cleanupExpiredMembers() {
long now = Instant.now().toEpochMilli();
redisTemplate.opsForZSet().removeRangeByScore(ConstantStr.DIFY_COUNTERRATELIMIT, 0, now);
}
/**
* 获取未过期的成员
*/
public Set<String> getValidMembers(String zsetKey) {
long now = Instant.now().toEpochMilli();
return redisTemplate.opsForZSet().rangeByScore(zsetKey, now, Double.MAX_VALUE);
}
public Long zCard(String zsetKey) {
return redisTemplate.opsForZSet().zCard(zsetKey);
}
}