|
|
@@ -0,0 +1,76 @@
|
|
|
+package com.adx.tencent.vivo.service;
|
|
|
+
|
|
|
+import com.adx.tencent.vivo.model.VivoTagEventRecord;
|
|
|
+import com.adx.tencent.vivo.store.VivoColdStore;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+import org.springframework.data.redis.connection.RedisConnection;
|
|
|
+import org.springframework.data.redis.connection.RedisStringCommands;
|
|
|
+import org.springframework.data.redis.core.StringRedisTemplate;
|
|
|
+import org.springframework.data.redis.core.types.Expiration;
|
|
|
+
|
|
|
+import java.nio.charset.StandardCharsets;
|
|
|
+import java.time.Duration;
|
|
|
+import java.util.LinkedHashMap;
|
|
|
+import java.util.LinkedHashSet;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.Set;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+public class VivoTagEventProfileSyncService {
|
|
|
+ private static final Logger log = LoggerFactory.getLogger(VivoTagEventProfileSyncService.class);
|
|
|
+
|
|
|
+ private final VivoColdStore coldStore;
|
|
|
+ private final StringRedisTemplate redis;
|
|
|
+ private final String keyPrefix;
|
|
|
+ private final Duration ttl;
|
|
|
+
|
|
|
+ public VivoTagEventProfileSyncService(VivoColdStore coldStore, StringRedisTemplate redis,
|
|
|
+ String keyPrefix, Duration ttl) {
|
|
|
+ this.coldStore = coldStore;
|
|
|
+ this.redis = redis;
|
|
|
+ this.keyPrefix = keyPrefix == null ? "" : keyPrefix;
|
|
|
+ this.ttl = ttl;
|
|
|
+ }
|
|
|
+
|
|
|
+ public int syncOnce() {
|
|
|
+ List<VivoTagEventRecord> rows = coldStore.listAllTagEventProfiles();
|
|
|
+ if (rows == null || rows.isEmpty()) {
|
|
|
+ log.info("[VivoTagEventProfileSync] 无数据可同步");
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+ Map<String, Set<String>> grouped = new LinkedHashMap<>();
|
|
|
+ for (VivoTagEventRecord row : rows) {
|
|
|
+ if (row.getTagId() == null || row.getTagId().isBlank()
|
|
|
+ || row.getCallbackProfile() == null || row.getCallbackProfile().isBlank()
|
|
|
+ || row.getEventType() == null || row.getEventType().isBlank()) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ grouped.computeIfAbsent(buildRedisKey(row.getTagId(), row.getCallbackProfile(), row.getBaiduAct()),
|
|
|
+ ignored -> new LinkedHashSet<>()).add(row.getEventType().trim());
|
|
|
+ }
|
|
|
+ long ttlMs = (ttl != null && !ttl.isZero()) ? ttl.toMillis() : 0;
|
|
|
+ redis.executePipelined((RedisConnection conn) -> {
|
|
|
+ for (Map.Entry<String, Set<String>> entry : grouped.entrySet()) {
|
|
|
+ byte[] key = entry.getKey().getBytes(StandardCharsets.UTF_8);
|
|
|
+ byte[] val = entry.getValue().stream().collect(Collectors.joining(","))
|
|
|
+ .getBytes(StandardCharsets.UTF_8);
|
|
|
+ if (ttlMs > 0) {
|
|
|
+ conn.stringCommands().set(key, val, Expiration.milliseconds(ttlMs),
|
|
|
+ RedisStringCommands.SetOption.UPSERT);
|
|
|
+ } else {
|
|
|
+ conn.stringCommands().set(key, val);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return null;
|
|
|
+ });
|
|
|
+ log.info("[VivoTagEventProfileSync] 同步完成,共 {} 条 vivo profile 回传方式写入 Redis, 聚合后 {} 个键",
|
|
|
+ rows.size(), grouped.size());
|
|
|
+ return rows.size();
|
|
|
+ }
|
|
|
+
|
|
|
+ private String buildRedisKey(String tagId, String profile, int baiduAct) {
|
|
|
+ return keyPrefix + "profile:" + tagId + ":" + profile + ":" + baiduAct;
|
|
|
+ }
|
|
|
+}
|