|
@@ -1,5 +1,9 @@
|
|
|
package com.tzld.piaoquan.recommend.feature.service;
|
|
package com.tzld.piaoquan.recommend.feature.service;
|
|
|
|
|
|
|
|
|
|
+import com.alibaba.fastjson.JSON;
|
|
|
|
|
+import com.ctrip.framework.apollo.model.ConfigChange;
|
|
|
|
|
+import com.ctrip.framework.apollo.model.ConfigChangeEvent;
|
|
|
|
|
+import com.ctrip.framework.apollo.spring.annotation.ApolloConfigChangeListener;
|
|
|
import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
|
|
import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
|
|
|
import com.google.common.base.Strings;
|
|
import com.google.common.base.Strings;
|
|
|
import com.tzld.piaoquan.recommend.feature.common.ThreadPoolFactory;
|
|
import com.tzld.piaoquan.recommend.feature.common.ThreadPoolFactory;
|
|
@@ -17,10 +21,8 @@ import org.springframework.beans.factory.annotation.Value;
|
|
|
import org.springframework.data.redis.core.RedisTemplate;
|
|
import org.springframework.data.redis.core.RedisTemplate;
|
|
|
import org.springframework.stereotype.Service;
|
|
import org.springframework.stereotype.Service;
|
|
|
|
|
|
|
|
-import java.util.ArrayList;
|
|
|
|
|
-import java.util.Collections;
|
|
|
|
|
-import java.util.List;
|
|
|
|
|
-import java.util.Optional;
|
|
|
|
|
|
|
+import javax.annotation.PostConstruct;
|
|
|
|
|
+import java.util.*;
|
|
|
import java.util.concurrent.*;
|
|
import java.util.concurrent.*;
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
@@ -36,9 +38,72 @@ public class FeatureV2Service {
|
|
|
@ApolloJsonValue("${dts.config:}")
|
|
@ApolloJsonValue("${dts.config:}")
|
|
|
private List<DTSConfig> newDtsConfigs;
|
|
private List<DTSConfig> newDtsConfigs;
|
|
|
|
|
|
|
|
- @Value("${feature.mget.batch.size:500}")
|
|
|
|
|
|
|
+ @Value("${feature.mget.batch.size:200}")
|
|
|
private Integer batchSize;
|
|
private Integer batchSize;
|
|
|
|
|
|
|
|
|
|
+ /**
|
|
|
|
|
+ * DTSConfig 缓存,按 tableName 索引,避免每次 redisKey() 都遍历列表
|
|
|
|
|
+ * 启动时通过 @PostConstruct 初始化,配置变更时通过 @ApolloConfigChangeListener 更新
|
|
|
|
|
+ */
|
|
|
|
|
+ private volatile Map<String, DTSConfig> dtsConfigCache = Collections.emptyMap();
|
|
|
|
|
+
|
|
|
|
|
+ @PostConstruct
|
|
|
|
|
+ public void init() {
|
|
|
|
|
+ rebuildDtsConfigCache();
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 重建 DTSConfig 缓存(从 newDtsConfigs 字段)
|
|
|
|
|
+ */
|
|
|
|
|
+ private void rebuildDtsConfigCache() {
|
|
|
|
|
+ rebuildDtsConfigCacheFromList(newDtsConfigs);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 重建 DTSConfig 缓存
|
|
|
|
|
+ * @param configs 配置列表
|
|
|
|
|
+ */
|
|
|
|
|
+ private void rebuildDtsConfigCacheFromList(List<DTSConfig> configs) {
|
|
|
|
|
+ if (configs == null || configs.isEmpty()) {
|
|
|
|
|
+ dtsConfigCache = Collections.emptyMap();
|
|
|
|
|
+ log.info("DTSConfig cache rebuilt, size=0");
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+ Map<String, DTSConfig> cache = new HashMap<>(configs.size());
|
|
|
|
|
+ for (DTSConfig config : configs) {
|
|
|
|
|
+ if (config.getOdps() != null && StringUtils.isNotBlank(config.getOdps().getTable())) {
|
|
|
|
|
+ cache.put(config.getOdps().getTable(), config);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ dtsConfigCache = cache;
|
|
|
|
|
+ log.info("DTSConfig cache rebuilt, size={}", cache.size());
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ /**
|
|
|
|
|
+ * 监听 Apollo 配置变更,自动重建缓存
|
|
|
|
|
+ * 手动解析新配置值,避免与 @ApolloJsonValue 自动注入产生竞态条件
|
|
|
|
|
+ */
|
|
|
|
|
+ @ApolloConfigChangeListener
|
|
|
|
|
+ public void onConfigChange(ConfigChangeEvent changeEvent) {
|
|
|
|
|
+ if (changeEvent.isChanged("dts.config")) {
|
|
|
|
|
+ ConfigChange change = changeEvent.getChange("dts.config");
|
|
|
|
|
+ String newValue = change.getNewValue();
|
|
|
|
|
+ log.info("dts.config changed, old={}, new={}", change.getOldValue(), newValue);
|
|
|
|
|
+
|
|
|
|
|
+ if (StringUtils.isNotBlank(newValue)) {
|
|
|
|
|
+ try {
|
|
|
|
|
+ List<DTSConfig> newConfigs = JSON.parseArray(newValue, DTSConfig.class);
|
|
|
|
|
+ rebuildDtsConfigCacheFromList(newConfigs);
|
|
|
|
|
+ } catch (Exception e) {
|
|
|
|
|
+ log.error("Failed to parse dts.config: {}", newValue, e);
|
|
|
|
|
+ }
|
|
|
|
|
+ } else {
|
|
|
|
|
+ dtsConfigCache = Collections.emptyMap();
|
|
|
|
|
+ log.info("dts.config is empty, cache cleared");
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
public MultiGetFeatureResponse multiGetFeature(MultiGetFeatureRequest request) {
|
|
public MultiGetFeatureResponse multiGetFeature(MultiGetFeatureRequest request) {
|
|
|
int keyCount = request.getFeatureKeyCount();
|
|
int keyCount = request.getFeatureKeyCount();
|
|
|
|
|
|
|
@@ -53,7 +118,7 @@ public class FeatureV2Service {
|
|
|
// 1. 生成Redis Key列表
|
|
// 1. 生成Redis Key列表
|
|
|
List<String> redisKeys = CommonCollectionUtils.toList(request.getFeatureKeyList(), this::redisKey);
|
|
List<String> redisKeys = CommonCollectionUtils.toList(request.getFeatureKeyList(), this::redisKey);
|
|
|
|
|
|
|
|
- int safeBatchSize = (batchSize == null || batchSize <= 0) ? 500 : batchSize;
|
|
|
|
|
|
|
+ int safeBatchSize = (batchSize == null || batchSize <= 0) ? 300 : batchSize;
|
|
|
|
|
|
|
|
// 2. 分批并行查询(必须保证返回值与 key 顺序一致)
|
|
// 2. 分批并行查询(必须保证返回值与 key 顺序一致)
|
|
|
List<BatchFuture> futures = new ArrayList<>();
|
|
List<BatchFuture> futures = new ArrayList<>();
|
|
@@ -120,15 +185,13 @@ public class FeatureV2Service {
|
|
|
|
|
|
|
|
// Note:写入和读取的key生成规则应保持一致
|
|
// Note:写入和读取的key生成规则应保持一致
|
|
|
private String redisKey(FeatureKeyProto fk) {
|
|
private String redisKey(FeatureKeyProto fk) {
|
|
|
-
|
|
|
|
|
- Optional<DTSConfig> optional = newDtsConfigs.stream()
|
|
|
|
|
- .filter(c -> c.getOdps() != null && StringUtils.equals(c.getOdps().getTable(), fk.getTableName()))
|
|
|
|
|
- .findFirst();
|
|
|
|
|
- if (!optional.isPresent()) {
|
|
|
|
|
|
|
+ // 使用缓存查找配置,O(1) 复杂度
|
|
|
|
|
+ // 缓存在启动时初始化,配置变更时通过 Apollo 监听器更新
|
|
|
|
|
+ DTSConfig config = dtsConfigCache.get(fk.getTableName());
|
|
|
|
|
+ if (config == null) {
|
|
|
log.error("table {} not config", fk.getTableName());
|
|
log.error("table {} not config", fk.getTableName());
|
|
|
return "";
|
|
return "";
|
|
|
}
|
|
}
|
|
|
- DTSConfig config = optional.get();
|
|
|
|
|
|
|
|
|
|
// Note:写入和读取的key生成规则应保持一致
|
|
// Note:写入和读取的key生成规则应保持一致
|
|
|
List<String> fields = config.getRedis().getKey();
|
|
List<String> fields = config.getRedis().getKey();
|