feat:特征服务优化

recommend-feature-service/src/main/java/com/tzld/piaoquan/recommend/feature/service/FeatureV2Service.java

@@ -43,94 +43,38 @@ public class FeatureV2Service {
     @ApolloJsonValue("${dts.config:}")
     private List<DTSConfig> newDtsConfigs;
 
+    @ApolloJsonValue("${dts.config.v2:}")
+    private List<DTSConfig> dtsConfigsV2;
+
     @Value("${feature.mget.batch.size:300}")
     private Integer batchSize;
 
-    @Value("${remove.base64.switch:false}")
-    private Boolean removeBase64Switch;
+    @Value("${feature.service.optimize.switch:false}")
+    private Boolean optimizeV3Switch;
 
     public MultiGetFeatureResponse multiGetFeature(MultiGetFeatureRequest request) {
         int keyCount = request.getFeatureKeyCount();
-
         if (keyCount == 0) {
             return MultiGetFeatureResponse.newBuilder()
                     .setResult(Result.newBuilder().setCode(1))
                     .build();
         }
 
-
-        // 移除Base64编解码的key
-        if (Boolean.TRUE.equals(removeBase64Switch)) {
-            return this.multiGetFeatureV2(request);
-        }
-
         long startTime = System.currentTimeMillis();
 
-        // 1. 生成Redis Key列表
-        List<String> redisKeys = CommonCollectionUtils.toList(request.getFeatureKeyList(), this::redisKey);
-
-        int safeBatchSize = (batchSize == null || batchSize <= 0) ? 300 : batchSize;
-
-        // 2. 分批并行查询（必须保证返回值与 key 顺序一致）
-        List<BatchFuture<String>> futures = new ArrayList<>();
-        for (int start = 0; start < redisKeys.size(); start += safeBatchSize) {
-            int end = Math.min(start + safeBatchSize, redisKeys.size());
-            List<String> batchKeys = redisKeys.subList(start, end);
-            Future<List<String>> future = ThreadPoolFactory.multiGetFeaturePool()
-                    .submit(() -> redisTemplate.opsForValue().multiGet(batchKeys));
-            futures.add(new BatchFuture<>(start, end - start, future));
+        List<String> redisKeys;
+        if (Boolean.TRUE.equals(optimizeV3Switch)) {
+            redisKeys = request.getFeatureKeyList().stream()
+                    .map(this::buildRedisKeyV2)
+                    .collect(Collectors.toList());
+        } else {
+            redisKeys = request.getFeatureKeyList().stream()
+                    .map(this::redisKey)
+                    .map(s -> String.format("%s:v2", s))
+                    .collect(Collectors.toList());
         }
 
-        // 3. 收集结果（整体最多等待约 2s；超时/异常的批次用 null 填充，避免错位）
-        List<String> allValues = new ArrayList<>(Collections.nCopies(keyCount, null));
-        long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(2000);
-        for (BatchFuture<String> bf : futures) {
-            long remainingNanos = deadlineNanos - System.nanoTime();
-            if (remainingNanos <= 0) {
-                bf.future.cancel(true);
-                continue;
-            }
-            try {
-                List<String> batchValues = bf.future.get(remainingNanos, TimeUnit.NANOSECONDS);
-                if (batchValues == null) {
-                    continue;
-                }
-                int copyLen = Math.min(bf.size, batchValues.size());
-                for (int i = 0; i < copyLen; i++) {
-                    allValues.set(bf.startIndex + i, batchValues.get(i));
-                }
-            } catch (TimeoutException e) {
-                bf.future.cancel(true);
-                log.warn("Batch mGet timeout, startIndex={}, size={}", bf.startIndex, bf.size);
-            } catch (Exception e) {
-                log.error("Batch mGet failed, startIndex={}, size={}", bf.startIndex, bf.size, e);
-            }
-        }
-
-        // 4. 解压缩
-        List<String> values = CommonCollectionUtils.toList(allValues, CompressionUtil::snappyDecompress);
-
-        // 5. 构建响应
-        MultiGetFeatureResponse.Builder builder = MultiGetFeatureResponse.newBuilder();
-        builder.setResult(Result.newBuilder().setCode(1));
-        for (int i = 0; i < keyCount; i++) {
-            builder.putFeature(request.getFeatureKeyList().get(i).getUniqueKey(),
-                    Strings.nullToEmpty(values.get(i)));
-        }
-        MultiGetFeatureResponse build = builder.build();
-        log.info("multiGetFeature, cost={}ms, keyCount={}", System.currentTimeMillis() - startTime, keyCount);
-        return build;
-    }
-
-    private MultiGetFeatureResponse multiGetFeatureV2(MultiGetFeatureRequest request) {
-        long startTime = System.currentTimeMillis();
-        int keyCount = request.getFeatureKeyCount();
-
-        // 1. 生成Redis Key列表, v2去除了base64编码 key格式也变了
-        List<String> redisKeys = request.getFeatureKeyList().stream()
-                .map(this::redisKey)
-                .map(s -> String.format("%s:v2", s))
-                .collect(Collectors.toList());
+        log.info("redisKey: {}", redisKeys);
 
         int safeBatchSize = (batchSize == null || batchSize <= 0) ? 300 : batchSize;
 
@@ -197,10 +141,17 @@ public class FeatureV2Service {
         }
     }
 
-    // Note：写入和读取的key生成规则应保持一致
     private String redisKey(FeatureKeyProto fk) {
+        return this.buildRedisKey(fk, newDtsConfigs);
+    }
+
+    private String buildRedisKeyV2(FeatureKeyProto fk) {
+        return this.buildRedisKey(fk, dtsConfigsV2);
+    }
 
-        Optional<DTSConfig> optional = newDtsConfigs.stream()
+    // Note：写入和读取的key生成规则应保持一致
+    private String buildRedisKey(FeatureKeyProto fk, List<DTSConfig> dtsConfigs){
+        Optional<DTSConfig> optional = dtsConfigs.stream()
                 .filter(c -> c.getOdps() != null && StringUtils.equals(c.getOdps().getTable(), fk.getTableName()))
                 .findFirst();
         if (!optional.isPresent()) {
@@ -224,5 +175,4 @@ public class FeatureV2Service {
         }
         return sb.toString();
     }
-
 }

recommend-feature-service/src/main/resources/application.yml

@@ -18,5 +18,5 @@ app:
 apollo:
   bootstrap:
     enabled: true
-    namespaces: application
+    namespaces: application,RD.Experiment
   cacheDir: /datalog/apollo-cache-dir