修改redis限制方法

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/RedisTestController.java

@@ -1,11 +1,9 @@
 package com.volvo.ai.analytic.center.controller;
 
 
-import com.volvo.ai.analytic.center.dto.req.AnalysisQueryReq;
+import com.alibaba.fastjson.JSONObject;
-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.ai.analytic.center.utils.ConstantStr;
+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;
@@ -14,15 +12,13 @@ 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.data.redis.core.ZSetOperations;
-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;
 
+import java.time.Instant;
 import java.util.Set;
-import java.util.concurrent.TimeUnit;
 
 
 @RestController()
@@ -41,49 +37,30 @@ public class RedisTestController {
 
     @Autowired
     private RedisTemplate<String, String> redisTemplate;
-
+    @Autowired
-
+    private RedisZSetUtil redisZSetUtil;
 
     @PostMapping("/testRedis")
-    @ApiOperation(value = "Ai解析结果查询")
+    public void test(@RequestBody String message) {
-    public AnalysisResp<Object> testRedis(@RequestBody String key) {
+        String zsetKey = "myZSet";
-        log.info("aiAnalyze queryRedis  : {}",key);
+        JSONObject json = JSONObject.parseObject(message);
+        String key = json.getString("key");
+        String value =json.getString("value");
+    // 添加成员并设置过期时间戳
+        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);
 
-        for (int i = 0; i < 10; i++){
+        // 获取未过期的成员
-            ZSetOperations zSetOperations = redisTemplate.opsForZSet();
+        Set<String> validMembers = redisZSetUtil.getValidMembers(zsetKey);
-            long currentMs = System.currentTimeMillis();
+        System.out.println("Valid members: " + validMembers);
-            zSetOperations.add(key, key+i, currentMs);
-
-            // set the expiration time for the code user
-            //redisTemplate.expire(key, 60, TimeUnit.SECONDS);
-
-            // remove the value that out of current window
-            zSetOperations.removeRangeByScore(key, 0, currentMs - 5 * 1000);
-
-            // check all available count
-            Long count = zSetOperations.zCard(key);
-            System.out.println("count:" + count);
-
-            Set<String> getredisSet = redisTemplate.opsForZSet().range(key, 0, -1);
-            System.out.println("getredisSet:" + getredisSet);
-            try {
-                TimeUnit.SECONDS.sleep(5);
-            } catch (InterruptedException e) {
-                throw new RuntimeException(e);
-            }
-            Long count2 = zSetOperations.zCard(key);
-            System.out.println("count2:" + count2);
-
-            Set<String> getredisSet2 = redisTemplate.opsForZSet().range(key, 0, -1);
-            System.out.println("getredisSet2:" + getredisSet2);
     }
 
-        return  AnalysisResp.success("ok");
-    }
     @PostMapping("/queryRedis")
     @ApiOperation(value = "queryRedis")
     public ResultMsg<Object> queryRedis(@RequestBody String key) {
-        log.info("aiAnalyze queryRedis  : {}",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);

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

@@ -2,7 +2,6 @@
 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.req.DiFyReq;
 import com.volvo.ai.analytic.center.entity.AiAnalysisErrors;
 import com.volvo.ai.analytic.center.service.AiAnalysisErrorsService;
@@ -10,6 +9,7 @@ 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.RedisZSetUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
@@ -18,10 +18,10 @@ 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.data.redis.core.ZSetOperations;
 import org.springframework.stereotype.Component;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.time.Instant;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -38,7 +38,7 @@ import java.util.concurrent.CompletableFuture;
 @RestController
 @RocketMQMessageListener(topic = "${rocketmq.consumer.analysisDify.topic}",consumerGroup = "${rocketmq.consumer.analysisDify.group}",
 		instanceName = "analysisDifyMqConsumer",
-		consumeThreadNumber = 40,
+		consumeThreadNumber = 1,
 		enableMsgTrace = true)
 public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt>  {
 
@@ -63,20 +63,19 @@ public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt>  {
 	@Autowired
 	private AiAnalysisErrorsService aiAnalysisErrorsService;
 
-	private final ObjectMapper objectMapper = new ObjectMapper();
-
 	@Autowired
 	private RedisTemplate<String, String> redisTemplate;
 
+	@Autowired
+	private RedisZSetUtil redisZSetUtil;
 	@Override
 	public void onMessage(MessageExt messageExt) {
 		log.info("analysisDifyMqConsumer 当前线程: {}, 线程ID: {}", Thread.currentThread().getName(), Thread.currentThread().getId());
 		long startTime = System.currentTimeMillis();
-		ZSetOperations zSetOperations = redisTemplate.opsForZSet();
+		Long count = redisZSetUtil.zCard(ConstantStr.DIFY_COUNTERRATELIMIT);
-		Long count = zSetOperations.zCard(ConstantStr.DIFY_COUNTERRATELIMIT);
+		log.info("redis计数数量: {}", count);
-		log.info("redis计数数量: " + count);
 		if(count>=difyLimit){
-			log.info("analysisDifyMqConsumer 请求dify超过基数 ：{}，稍后请求: " + redisCounterRateLimiter.getCurrentCount(ConstantStr.DIFY_COUNTERRATELIMIT));
+			log.info("analysisDifyMqConsumer 请求dify超过基数 ：{}，稍后请求: " +  redisZSetUtil.getValidMembers(ConstantStr.DIFY_COUNTERRATELIMIT));
 			throw new RuntimeException("请求dify超过基数 " );
 		}
 		String message = new String(messageExt.getBody());
@@ -84,12 +83,10 @@ public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt>  {
 		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秒后过期
-		zSetOperations.add(ConstantStr.DIFY_COUNTERRATELIMIT, aiAnalysisRequestId, startTime);
+		redisZSetUtil.addWithExpire(ConstantStr.DIFY_COUNTERRATELIMIT, aiAnalysisRequestId, expireTime);
-		zSetOperations.removeRangeByScore(ConstantStr.DIFY_COUNTERRATELIMIT, 0, startTime - (expire * 1000));
 		try {
  			CompletableFuture<JSONObject> future = diFyService.asyncExecuteDifyFlow(difyReq);
-			//			JSONObject json = future.get();
 			future.thenAccept(result ->  {
 				log.info("异步处理asyncExecuteDifyFlow完成aiAnalysisRequestId: {},处理结果：{}", aiAnalysisRequestId, result);
 				JSONObject data = result.getJSONObject("data");

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisZSetUtil.java

@@ -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);
+    }
+}