修改为10分钟

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

@@ -33,6 +33,7 @@ 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(10);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -62,27 +63,24 @@ public class HonorHotStore {
     public void recordBid(HonorBidRecord record) {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
-            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
-            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
+            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                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[] 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
+                );
                 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));
@@ -154,6 +152,10 @@ 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,6 +35,7 @@ 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(10);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -64,30 +65,27 @@ public class KuaishouHotStore {
     public void recordBid(KuaishouBidRecord record) {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
-            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
-            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
+            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                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
-                        );
-                    }
+                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
+                    );
                 }
                 Map<byte[], byte[]> body = new LinkedHashMap<>();
                 body.put("type".getBytes(StandardCharsets.UTF_8), "bid".getBytes(StandardCharsets.UTF_8));
@@ -146,6 +144,10 @@ 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,6 +36,7 @@ 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(10);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -65,30 +66,27 @@ public class OppoHotStore {
     public void recordBid(OppoBidRecord record) {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
-            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
-            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
+            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                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
-                        );
-                    }
+                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
+                    );
                 }
                 Map<byte[], byte[]> body = new LinkedHashMap<>();
                 body.put("type".getBytes(StandardCharsets.UTF_8), "bid".getBytes(StandardCharsets.UTF_8));
@@ -324,6 +322,10 @@ 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,6 +27,7 @@ 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(10);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -59,40 +60,39 @@ public class RedisHotStore {
     // ─── EventRecorder ───────────────────────────────────────────────────────
 
     /**
-     * 对应 Go RecordBid：曝光事件只写 Stream，其他事件写热缓存并写 Stream。
+     * 对应 Go RecordBid：
+     *   Pipeline: SET bid:{qk} + SET media:{media}:{traceId} + XADD
      */
     public void recordBid(BidRecord record) {
         if (record.getCreatedAt() == null) {
             record.setCreatedAt(Instant.now());
         }
         try {
-            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
-            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
+            // Redis KEY 只存关键字段（瘦身），降低内存占用
+            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
             // Stream 存全量数据，供 ColdWorker 持久化到 TiDB
             String fullPayload = objectMapper.writeValueAsString(record);
 
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                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
-                        );
-                    }
+                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
+                    );
                 }
                 // XADD 全量数据到 Stream
                 byte[] streamKey = stream.getBytes(StandardCharsets.UTF_8);
@@ -175,6 +175,10 @@ 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,6 +35,7 @@ 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(10);
     private static final Duration DEFAULT_CONVERSION_SEEN_TTL = Duration.ofDays(7);
     private static final Duration DEFAULT_DEDUCTION_COUNTER_TTL = Duration.ofDays(7);
 
@@ -64,30 +65,27 @@ public class VivoHotStore {
     public void recordBid(VivoBidRecord record) {
         if (record.getCreatedAt() == null) record.setCreatedAt(Instant.now());
         try {
-            boolean cacheHot = !"impression".equalsIgnoreCase(record.getEventType());
-            String hotPayload = cacheHot ? objectMapper.writeValueAsString(record.toHotRecord()) : null;
+            String hotPayload = objectMapper.writeValueAsString(record.toHotRecord());
             String fullPayload = objectMapper.writeValueAsString(record);
             redis.executePipelined((org.springframework.data.redis.connection.RedisConnection conn) -> {
                 trimStreamIfNeeded(conn);
-                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
-                        );
-                    }
+                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
+                    );
                 }
                 Map<byte[], byte[]> body = new LinkedHashMap<>();
                 body.put("type".getBytes(StandardCharsets.UTF_8), "bid".getBytes(StandardCharsets.UTF_8));
@@ -163,6 +161,10 @@ 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,10 +51,17 @@ class OppoHotStoreTest {
 
     @Test
     @SuppressWarnings({"rawtypes", "unchecked"})
-    void impressionSkipsHotCacheButStreamKeepsFullEvent() throws Exception {
+    void impressionCachesAreSmallAndExpireInTenMinutesButStreamKeepsFullEvent() throws Exception {
         OppoBidRecord record = record();
         store(Duration.ofHours(36)).recordBid(record);
-        verifyNoInteractions(stringCommands);
+        JsonNode hot = cachePayload(Duration.ofMinutes(10));
+        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);
 
         ArgumentCaptor<MapRecord<byte[], byte[], byte[]>> event = ArgumentCaptor.forClass(MapRecord.class);
         verify(streamCommands).xAdd(event.capture());