增加任务频次

src/main/java/com/adx/tencent/honor/store/HonorHotStore.java

@@ -33,7 +33,6 @@ import java.util.concurrent.ConcurrentHashMap;
 public class HonorHotStore {
 
     private static final Logger log = LoggerFactory.getLogger(HonorHotStore.class);
-    private static final Duration IMPRESSION_BID_TTL = Duration.ofMinutes(30);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -63,24 +62,27 @@ public class HonorHotStore {
     public void recordBid(HonorBidRecord record) {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
-            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
+            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
+            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
-                long ttlMillis = bidTtlFor(record.getEventType()).toMillis();
-                conn.stringCommands().set(
-                        bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
-                        hotBytes,
-                        org.springframework.data.redis.core.types.Expiration.milliseconds(ttlMillis),
-                        org.springframework.data.redis.connection.RedisStringCommands.SetOption.UPSERT
-                );
-                conn.stringCommands().set(
-                        mediaTraceKey(record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
-                        hotBytes,
-                        org.springframework.data.redis.core.types.Expiration.milliseconds(ttlMillis),
-                        org.springframework.data.redis.connection.RedisStringCommands.SetOption.UPSERT
-                );
+                if (cacheHot) {
+                    byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
+                    long ttlMillis = resolvedBidTtl().toMillis();
+                    conn.stringCommands().set(
+                            bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
+                            hotBytes,
+                            org.springframework.data.redis.core.types.Expiration.milliseconds(ttlMillis),
+                            org.springframework.data.redis.connection.RedisStringCommands.SetOption.UPSERT
+                    );
+                    conn.stringCommands().set(
+                            mediaTraceKey(record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
+                            hotBytes,
+                            org.springframework.data.redis.core.types.Expiration.milliseconds(ttlMillis),
+                            org.springframework.data.redis.connection.RedisStringCommands.SetOption.UPSERT
+                    );
+                }
                 byte[] streamKey = stream.getBytes(StandardCharsets.UTF_8);
                 Map<byte[], byte[]> body = new LinkedHashMap<>();
                 body.put("type".getBytes(StandardCharsets.UTF_8), "bid".getBytes(StandardCharsets.UTF_8));
@@ -152,10 +154,6 @@ public class HonorHotStore {
         return parseBidRecord(payload);
     }
 
-    private Duration bidTtlFor(String eventType) {
-        return "impression".equalsIgnoreCase(eventType) ? IMPRESSION_BID_TTL : resolvedBidTtl();
-    }
-
     private Duration resolvedBidTtl() {
         return (bidTtl == null || bidTtl.isZero()) ? Duration.ofHours(36) : bidTtl;
     }

src/main/java/com/adx/tencent/kuaishou/store/KuaishouHotStore.java

@@ -35,7 +35,6 @@ import java.util.concurrent.ConcurrentHashMap;
 public class KuaishouHotStore {
 
     private static final Logger log = LoggerFactory.getLogger(KuaishouHotStore.class);
-    private static final Duration IMPRESSION_BID_TTL = Duration.ofMinutes(30);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -65,27 +64,30 @@ public class KuaishouHotStore {
     public void recordBid(KuaishouBidRecord record) {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
-            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
+            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
+            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
-                long ttlMillis = bidTtlFor(record.getEventType()).toMillis();
-                if (record.getQk() != null && !record.getQk().isBlank()) {
-                    conn.stringCommands().set(
-                            bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
-                            hotBytes,
-                            Expiration.milliseconds(ttlMillis),
-                            RedisStringCommands.SetOption.UPSERT
-                    );
-                }
-                if (record.getMediaTraceId() != null && !record.getMediaTraceId().isBlank()) {
-                    conn.stringCommands().set(
-                            mediaTraceKey(record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
-                            hotBytes,
-                            Expiration.milliseconds(ttlMillis),
-                            RedisStringCommands.SetOption.UPSERT
-                    );
+                if (cacheHot) {
+                    byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
+                    long ttlMillis = resolvedBidTtl().toMillis();
+                    if (record.getQk() != null && !record.getQk().isBlank()) {
+                        conn.stringCommands().set(
+                                bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
+                                hotBytes,
+                                Expiration.milliseconds(ttlMillis),
+                                RedisStringCommands.SetOption.UPSERT
+                        );
+                    }
+                    if (record.getMediaTraceId() != null && !record.getMediaTraceId().isBlank()) {
+                        conn.stringCommands().set(
+                                mediaTraceKey(record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
+                                hotBytes,
+                                Expiration.milliseconds(ttlMillis),
+                                RedisStringCommands.SetOption.UPSERT
+                        );
+                    }
                 }
                 Map<byte[], byte[]> body = new LinkedHashMap<>();
                 body.put("type".getBytes(StandardCharsets.UTF_8), "bid".getBytes(StandardCharsets.UTF_8));
@@ -144,10 +146,6 @@ public class KuaishouHotStore {
         return parseBidRecord(payload);
     }
 
-    private Duration bidTtlFor(String eventType) {
-        return "impression".equalsIgnoreCase(eventType) ? IMPRESSION_BID_TTL : resolvedBidTtl();
-    }
-
     private Duration resolvedBidTtl() {
         return (bidTtl == null || bidTtl.isZero()) ? Duration.ofHours(36) : bidTtl;
     }

src/main/java/com/adx/tencent/oppo/store/OppoHotStore.java

@@ -36,7 +36,6 @@ import java.util.concurrent.ConcurrentHashMap;
 public class OppoHotStore {
 
     private static final Logger log = LoggerFactory.getLogger(OppoHotStore.class);
-    private static final Duration IMPRESSION_BID_TTL = Duration.ofMinutes(30);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -66,27 +65,30 @@ public class OppoHotStore {
     public void recordBid(OppoBidRecord record) {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
-            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
+            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
+            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
-                long ttlMillis = bidTtlFor(record.getEventType()).toMillis();
-                if (record.getQk() != null && !record.getQk().isBlank()) {
-                    conn.stringCommands().set(
-                            bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
-                            hotBytes,
-                            Expiration.milliseconds(ttlMillis),
-                            RedisStringCommands.SetOption.UPSERT
-                    );
-                }
-                if (record.getMediaTraceId() != null && !record.getMediaTraceId().isBlank()) {
-                    conn.stringCommands().set(
-                            mediaTraceKey(record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
-                            hotBytes,
-                            Expiration.milliseconds(ttlMillis),
-                            RedisStringCommands.SetOption.UPSERT
-                    );
+                if (cacheHot) {
+                    byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
+                    long ttlMillis = resolvedBidTtl().toMillis();
+                    if (record.getQk() != null && !record.getQk().isBlank()) {
+                        conn.stringCommands().set(
+                                bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
+                                hotBytes,
+                                Expiration.milliseconds(ttlMillis),
+                                RedisStringCommands.SetOption.UPSERT
+                        );
+                    }
+                    if (record.getMediaTraceId() != null && !record.getMediaTraceId().isBlank()) {
+                        conn.stringCommands().set(
+                                mediaTraceKey(record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
+                                hotBytes,
+                                Expiration.milliseconds(ttlMillis),
+                                RedisStringCommands.SetOption.UPSERT
+                        );
+                    }
                 }
                 Map<byte[], byte[]> body = new LinkedHashMap<>();
                 body.put("type".getBytes(StandardCharsets.UTF_8), "bid".getBytes(StandardCharsets.UTF_8));
@@ -322,10 +324,6 @@ public class OppoHotStore {
         conn.streamCommands().xTrim(stream.getBytes(StandardCharsets.UTF_8), streamMaxLen);
     }
 
-    private Duration bidTtlFor(String eventType) {
-        return "impression".equalsIgnoreCase(eventType) ? IMPRESSION_BID_TTL : resolvedBidTtl();
-    }
-
     private Duration resolvedBidTtl() {
         return (bidTtl == null || bidTtl.isZero()) ? Duration.ofHours(36) : bidTtl;
     }

src/main/java/com/adx/tencent/storage/RedisHotStore.java

@@ -27,7 +27,6 @@ public class RedisHotStore {
 
     private static final Logger log = LoggerFactory.getLogger(RedisHotStore.class);
     public static final String BID_NOT_FOUND_MSG = "bid not found";
-    private static final Duration IMPRESSION_BID_TTL = Duration.ofMinutes(30);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -60,39 +59,40 @@ public class RedisHotStore {
     // ─── EventRecorder ───────────────────────────────────────────────────────
 
     /**
-     * 对应 Go RecordBid：
-     *   Pipeline: SET bid:{qk} + SET media:{media}:{traceId} + XADD
+     * 对应 Go RecordBid：曝光事件只写 Stream，其他事件写热缓存并写 Stream。
      */
     public void recordBid(BidRecord record) {
         if (record.getCreatedAt() == null) {
             record.setCreatedAt(Instant.now());
         }
         try {
-            // Redis KEY 只存关键字段（瘦身），降低内存占用
-            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
+            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
+            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
             // Stream 存全量数据，供 ColdWorker 持久化到 TiDB
             String fullPayload = objectMapper.writeValueAsString(record);
 
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
-                long ttlMillis = bidTtlFor(record.getEventType()).toMillis();
-                if (record.getQk() != null && !record.getQk().isEmpty()) {
-                    conn.stringCommands().set(
-                            bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
-                            hotBytes,
-                            org.springframework.data.redis.core.types.Expiration.milliseconds(ttlMillis),
-                            org.springframework.data.redis.connection.RedisStringCommands.SetOption.UPSERT
-                    );
-                }
-                if (record.getMedia() != null && !record.getMedia().isEmpty()
-                        && record.getMediaTraceId() != null && !record.getMediaTraceId().isEmpty()) {
-                    conn.stringCommands().set(
-                            mediaTraceKey(record.getMedia(), record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
-                            hotBytes,
-                            org.springframework.data.redis.core.types.Expiration.milliseconds(ttlMillis),
-                            org.springframework.data.redis.connection.RedisStringCommands.SetOption.UPSERT
-                    );
+                if (cacheHot) {
+                    byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
+                    long ttlMillis = bidTtl.toMillis();
+                    if (record.getQk() != null && !record.getQk().isEmpty()) {
+                        conn.stringCommands().set(
+                                bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
+                                hotBytes,
+                                org.springframework.data.redis.core.types.Expiration.milliseconds(ttlMillis),
+                                org.springframework.data.redis.connection.RedisStringCommands.SetOption.UPSERT
+                        );
+                    }
+                    if (record.getMedia() != null && !record.getMedia().isEmpty()
+                            && record.getMediaTraceId() != null && !record.getMediaTraceId().isEmpty()) {
+                        conn.stringCommands().set(
+                                mediaTraceKey(record.getMedia(), record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
+                                hotBytes,
+                                org.springframework.data.redis.core.types.Expiration.milliseconds(ttlMillis),
+                                org.springframework.data.redis.connection.RedisStringCommands.SetOption.UPSERT
+                        );
+                    }
                 }
                 // XADD 全量数据到 Stream
                 byte[] streamKey = stream.getBytes(StandardCharsets.UTF_8);
@@ -175,10 +175,6 @@ public class RedisHotStore {
         return null;
     }
 
-    private Duration bidTtlFor(String eventType) {
-        return "impression".equalsIgnoreCase(eventType) ? IMPRESSION_BID_TTL : bidTtl;
-    }
-
     public Map<String, BidRecord> findBidsByQks(List<String> qks) {
         if (qks == null || qks.isEmpty()) return Collections.emptyMap();
         Map<String, BidRecord> result = new HashMap<>();

src/main/java/com/adx/tencent/vivo/store/VivoHotStore.java

@@ -35,7 +35,6 @@ import java.util.concurrent.ConcurrentHashMap;
 public class VivoHotStore {
 
     private static final Logger log = LoggerFactory.getLogger(VivoHotStore.class);
-    private static final Duration IMPRESSION_BID_TTL = Duration.ofMinutes(30);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -65,27 +64,30 @@ public class VivoHotStore {
     public void recordBid(VivoBidRecord record) {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
-            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
+            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
+            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
-                long ttlMillis = bidTtlFor(record.getEventType()).toMillis();
-                if (record.getQk() != null && !record.getQk().isBlank()) {
-                    conn.stringCommands().set(
-                            bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
-                            hotBytes,
-                            Expiration.milliseconds(ttlMillis),
-                            RedisStringCommands.SetOption.UPSERT
-                    );
-                }
-                if (record.getMediaTraceId() != null && !record.getMediaTraceId().isBlank()) {
-                    conn.stringCommands().set(
-                            mediaTraceKey(record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
-                            hotBytes,
-                            Expiration.milliseconds(ttlMillis),
-                            RedisStringCommands.SetOption.UPSERT
-                    );
+                if (cacheHot) {
+                    byte[] hotBytes = hotPayload.getBytes(StandardCharsets.UTF_8);
+                    long ttlMillis = resolvedBidTtl().toMillis();
+                    if (record.getQk() != null && !record.getQk().isBlank()) {
+                        conn.stringCommands().set(
+                                bidKey(record.getQk()).getBytes(StandardCharsets.UTF_8),
+                                hotBytes,
+                                Expiration.milliseconds(ttlMillis),
+                                RedisStringCommands.SetOption.UPSERT
+                        );
+                    }
+                    if (record.getMediaTraceId() != null && !record.getMediaTraceId().isBlank()) {
+                        conn.stringCommands().set(
+                                mediaTraceKey(record.getMediaTraceId()).getBytes(StandardCharsets.UTF_8),
+                                hotBytes,
+                                Expiration.milliseconds(ttlMillis),
+                                RedisStringCommands.SetOption.UPSERT
+                        );
+                    }
                 }
                 Map<byte[], byte[]> body = new LinkedHashMap<>();
                 body.put("type".getBytes(StandardCharsets.UTF_8), "bid".getBytes(StandardCharsets.UTF_8));
@@ -161,10 +163,6 @@ public class VivoHotStore {
         return parseBidRecord(payload);
     }
 
-    private Duration bidTtlFor(String eventType) {
-        return "impression".equalsIgnoreCase(eventType) ? IMPRESSION_BID_TTL : resolvedBidTtl();
-    }
-
     private Duration resolvedBidTtl() {
         return (bidTtl == null || bidTtl.isZero()) ? Duration.ofHours(36) : bidTtl;
     }

src/test/java/com/adx/tencent/oppo/store/OppoHotStoreTest.java

@@ -51,17 +51,10 @@ class OppoHotStoreTest {
 
     @Test
     @SuppressWarnings({"rawtypes", "unchecked"})
-    void impressionCachesAreSmallAndExpireInThirtyMinutesButStreamKeepsFullEvent() throws Exception {
+    void impressionSkipsHotCacheButStreamKeepsFullEvent() throws Exception {
         OppoBidRecord record = record();
         store(Duration.ofHours(36)).recordBid(record);
-        JsonNode hot = cachePayload(Duration.ofMinutes(30));
-        assertEquals("click-url", hot.get("clickUrls").get(0).asText());
-        assertEquals("store-url", hot.get("appStoreLink").asText());
-        assertFalse(hot.has("showUrls"));
-        assertFalse(hot.has("price"));
-        assertFalse(hot.has("createdAt"));
-        assertFalse(hot.get("mediaParams").has("unused"));
-        assertCallbackFields(hot);
+        verifyNoInteractions(stringCommands);
 
         ArgumentCaptor<MapRecord<byte[], byte[], byte[]>> event = ArgumentCaptor.forClass(MapRecord.class);
         verify(streamCommands).xAdd(event.capture());