ItemIndexService.java

1package com.reco.offline.index;23import java.util.*;4import java.util.concurrent.*;5import java.util.concurrent.atomic.AtomicReference;67/**8 * 物品索引服务 — 完整可运行示例9 *10 * 包含：11 *  1. DualBufferIndex        - 双 Buffer 全量索引发布（书中 code_dual_buffer_publish）12 *  2. OrderedInvertedIndex   - 有序倒排索引（书中 code_ordered_inverted_index）13 *  3. RealtimeIndexUpdater   - 实时增量更新（书中 code_realtime_index_update）14 *15 * [MOCK] 所有持久化操作使用内存存储；16 *        向量索引替换为 Faiss 或专有 ANN 库。17 */18public class ItemIndexService {1920    // ─────────────────────────────────────────────21    //  数据结构22    // ─────────────────────────────────────────────2324    public static class ItemFeature {25        public final long itemId;26        public final double score;27        public final Map<String, String> attributes;2829        public ItemFeature(long itemId, double score, Map<String, String> attributes) {30            this.itemId = itemId;31            this.score = score;32            this.attributes = Map.copyOf(attributes);33        }34    }3536    /** 倒排索引节点（itemId + 排序分） */37    public static class InvertedEntry implements Comparable<InvertedEntry> {38        public final long itemId;39        public final double sortScore;4041        public InvertedEntry(long itemId, double sortScore) {42            this.itemId = itemId;43            this.sortScore = sortScore;44        }4546        @Override47        public int compareTo(InvertedEntry o) {48            // 降序排列49            return Double.compare(o.sortScore, this.sortScore);50        }51    }5253    // ─────────────────────────────────────────────54    //  双 Buffer 全量索引发布55    //  对应书中 code_dual_buffer_publish56    // ─────────────────────────────────────────────5758    public static class DualBufferIndex {59        /** 当前读 Buffer（原子引用实现零拷贝切换） */60        private final AtomicReference<Map<Long, ItemFeature>> readBuffer =61                new AtomicReference<>(new ConcurrentHashMap<>());62        /** 写 Buffer */63        private volatile Map<Long, ItemFeature> writeBuffer = new ConcurrentHashMap<>();6465        /**66         * 全量加载新索引到写 Buffer，然后原子切换。67         * 调用期间读请求仍可访问旧 Buffer，无锁等待。68         */69        public void reload(List<ItemFeature> newData) {70            Map<Long, ItemFeature> newBuffer = new ConcurrentHashMap<>();71            for (ItemFeature f : newData) {72                newBuffer.put(f.itemId, f);73            }74            writeBuffer = newBuffer;75            // 原子切换：旧 buffer 由 GC 回收76            readBuffer.set(newBuffer);77            System.out.println("[DualBufferIndex] reloaded items=" + newBuffer.size());78        }7980        /** 并发安全的读取 */81        public ItemFeature get(long itemId) {82            return readBuffer.get().get(itemId);83        }8485        /** 实时插入或更新单个物品文档（供 RealtimeIndexUpdater 调用） */86        public void upsert(ItemFeature doc) {87            readBuffer.get().put(doc.itemId, doc);88        }8990        /** 从正排删除单个物品（供 RealtimeIndexUpdater 调用） */91        public void remove(long itemId) {92            readBuffer.get().remove(itemId);93        }9495        public int size() {96            return readBuffer.get().size();97        }98    }99100    // ─────────────────────────────────────────────101    //  有序倒排索引102    //  对应书中 code_ordered_inverted_index103    // ─────────────────────────────────────────────104105    public static class OrderedInvertedIndex {106        /** tag -> sorted list of InvertedEntry */107        private final Map<String, TreeSet<InvertedEntry>> invertedMap = new ConcurrentHashMap<>();108        /** itemId -> set of tags */109        private final Map<Long, Set<String>> forwardMap = new ConcurrentHashMap<>();110111        private static final int MAX_PER_TAG = 10_000;112113        public void upsert(long itemId, double score, Set<String> tags) {114            Set<String> oldTags = forwardMap.get(itemId);115            // 移除旧标签中该 item 的倒排条目116            if (oldTags != null) {117                for (String tag : oldTags) {118                    removeFromInverted(tag, itemId);119                }120            }121            // 插入新条目122            for (String tag : tags) {123                TreeSet<InvertedEntry> bucket = invertedMap.computeIfAbsent(124                        tag, k -> new TreeSet<>());125                bucket.add(new InvertedEntry(itemId, score));126                // 超出上限时移除分数最低的127                while (bucket.size() > MAX_PER_TAG) {128                    bucket.pollLast();129                }130            }131            forwardMap.put(itemId, new HashSet<>(tags));132        }133134        public void remove(long itemId) {135            Set<String> tags = forwardMap.remove(itemId);136            if (tags != null) {137                for (String tag : tags) removeFromInverted(tag, itemId);138            }139        }140141        private void removeFromInverted(String tag, long itemId) {142            TreeSet<InvertedEntry> bucket = invertedMap.get(tag);143            if (bucket != null) {144                bucket.removeIf(e -> e.itemId == itemId);145            }146        }147148        /** 按分数降序返回 topK 结果 */149        public List<InvertedEntry> query(String tag, int topK) {150            TreeSet<InvertedEntry> bucket = invertedMap.get(tag);151            if (bucket == null || bucket.isEmpty()) return List.of();152            List<InvertedEntry> result = new ArrayList<>();153            int count = 0;154            for (InvertedEntry e : bucket) {155                if (count++ >= topK) break;156                result.add(e);157            }158            return result;159        }160161        public int tagSize(String tag) {162            TreeSet<InvertedEntry> b = invertedMap.get(tag);163            return b == null ? 0 : b.size();164        }165    }166167    // ─────────────────────────────────────────────168    //  实时增量索引更新（Kafka Consumer）169    //  对应书中 code_realtime_index_update170    // ─────────────────────────────────────────────171172    /**173     * 索引更新触发消息（来自 Kafka 主题 index-update-trigger）。174     * 消息只携带物品 ID、变化原因和产生时间，不包含完整特征；175     * 消费者收到后需重新读取当前文档并重建索引。176     */177    public static class IndexUpdateTrigger {178        public final long itemId;179        public final String reason;   // e.g. "CONTENT_UPDATED", "INTERACTION_CHANGED", "OFFLINE"180        public final long eventTimeMs;181182        public IndexUpdateTrigger(long itemId, String reason, long eventTimeMs) {183            this.itemId = itemId;184            this.reason = reason;185            this.eventTimeMs = eventTimeMs;186        }187    }188189    /**190     * 实时索引更新 Kafka Consumer。191     * [MOCK] MockKafkaConsumer 为内存队列；工业落地时替换为真实 KafkaConsumer。192     * consume() 正常返回后由外部 Kafka 框架提交位点；异常必须向上传播，不可吞掉。193     */194    public static class RealtimeIndexUpdater {195        private final DualBufferIndex forwardIndex;196        private final OrderedInvertedIndex invertedIndex;197        // [MOCK] 模拟 Kafka Consumer 拉取触发消息198        private final Queue<List<IndexUpdateTrigger>> mockKafkaQueue = new ArrayDeque<>();199200        public RealtimeIndexUpdater(DualBufferIndex forwardIndex,201                                    OrderedInvertedIndex invertedIndex) {202            this.forwardIndex  = forwardIndex;203            this.invertedIndex = invertedIndex;204        }205206        /**207         * 消费一批 IndexUpdateTrigger 消息并同步更新正排与倒排索引。208         * 1. 提取并去重本批物品 ID209         * 2. 重新加载每个物品的当前特征文档（模拟 RPC 读取）210         * 3. 不可见/已下线物品 → 同时从正排和倒排删除211         * 4. 可见物品 → 写入正排，并更新倒排212         * 异常直接向上传播，调用方（Kafka 框架）负责提交或回退位点。213         */214        public void consume(List<IndexUpdateTrigger> triggers) {215            // Step 1: 提取并去重物品 ID216            Set<Long> itemIds = new LinkedHashSet<>();217            for (IndexUpdateTrigger t : triggers) {218                if (t.itemId > 0) itemIds.add(t.itemId);219            }220            if (itemIds.isEmpty()) return;221222            // Step 2: 批量重新加载当前特征文档（mock: 从 forwardIndex 读取已有文档）223            for (long itemId : itemIds) {224                ItemFeature doc = loadCurrentDoc(itemId);225226                if (doc == null || isOffline(doc)) {227                    // Step 3: 不可见或已下线 → 从正排和倒排同时删除228                    removeFromForwardIndex(itemId);229                    invertedIndex.remove(itemId);230                    System.out.printf("[RealtimeIndexUpdater] deleted itemId=%d%n", itemId);231                } else {232                    // Step 4: 写入正排，并更新倒排233                    upsertForwardIndex(doc);234                    Set<String> tags = deriveTags(doc.score, doc.attributes);235                    invertedIndex.upsert(doc.itemId, doc.score, tags);236                    System.out.printf("[RealtimeIndexUpdater] updated itemId=%d score=%.4f tags=%s%n",237                            doc.itemId, doc.score, tags);238                }239            }240        }241242        /**243         * [MOCK] 模拟从上游 RPC / Redis 重新读取物品当前完整文档。244         * 工业落地时调用内容理解服务、状态服务和统计服务，聚合为完整文档。245         */246        private ItemFeature loadCurrentDoc(long itemId) {247            // 先查正排中已有文档，作为 mock 数据源248            ItemFeature existing = forwardIndex.get(itemId);249            if (existing != null) return existing;250            // 若正排中不存在，模拟新增物品的特征文档251            return new ItemFeature(itemId, 0.5, Map.of("category", "general"));252        }253254        /** [MOCK] 判断物品是否已下线 */255        private boolean isOffline(ItemFeature doc) {256            return "offline".equalsIgnoreCase(doc.attributes.getOrDefault("status", ""));257        }258259        /** 写入正排索引 */260        private void upsertForwardIndex(ItemFeature doc) {261            forwardIndex.upsert(doc);262        }263264        /** 从正排索引删除 */265        private void removeFromForwardIndex(long itemId) {266            forwardIndex.remove(itemId);267        }268269        private Set<String> deriveTags(double score, Map<String, String> attributes) {270            Set<String> tags = new HashSet<>();271            String category = attributes.getOrDefault("category", "");272            if (!category.isBlank()) tags.add("cat:" + category);273            String author = attributes.getOrDefault("author_id", "");274            if (!author.isBlank()) tags.add("author:" + author);275            if (score > 0.8) tags.add("top_score");276            return tags;277        }278279        /** 向 mock Kafka 队列注入触发消息（测试用） */280        public void mockPush(List<IndexUpdateTrigger> triggers) {281            mockKafkaQueue.offer(triggers);282        }283284        /** 拉取并消费一批（测试用） */285        public void pollAndConsume() {286            List<IndexUpdateTrigger> batch = mockKafkaQueue.poll();287            if (batch != null) consume(batch);288        }289    }290    291292    // ─────────────────────────────────────────────293    //  离线全量索引构建入口294    //  对应书中 code_offline_runner295    // ─────────────────────────────────────────────296297    public static class OfflineRun {298        public final String id;299        public final java.time.Instant inputCutoff;300        public final String outputVersion;301        public OfflineRun(String id, java.time.Instant inputCutoff, String outputVersion) {302            this.id            = id;303            this.inputCutoff   = inputCutoff;304            this.outputVersion = outputVersion;305        }306    }307308    /**309     * 离线全量索引构建器 — 对应书中 code_offline_runner。310     * [MOCK] 从快照读取物品特征，填充双 Buffer 后原子切换；311     *        工业落地时快照数据从 Hive / HDFS 读取，312     *        通过 RPC 通知在线服务刷新索引版本。313     */314    public static class OfflineRunner {315        private final DualBufferIndex forwardIndex;316        private final OrderedInvertedIndex invertedIndex;317318        public OfflineRunner(DualBufferIndex forwardIndex,319                             OrderedInvertedIndex invertedIndex) {320            this.forwardIndex  = forwardIndex;321            this.invertedIndex = invertedIndex;322        }323324        public OfflineRun run(List<ItemFeature> snapshot) {325            String runId = "run-" + System.currentTimeMillis();326            java.time.Instant cutoff = java.time.Instant.now();327            // 1. 全量写入写 Buffer 并原子切换328            forwardIndex.reload(snapshot);329            // 2. 重建倒排索引330            for (ItemFeature f : snapshot) {331                Set<String> tags = deriveTags(f.score, f.attributes);332                invertedIndex.upsert(f.itemId, f.score, tags);333            }334            String version = "v" + cutoff.getEpochSecond();335            System.out.printf("[OfflineRunner] run=%s version=%s items=%d%n",336                    runId, version, snapshot.size());337            return new OfflineRun(runId, cutoff, version);338        }339340        private Set<String> deriveTags(double score, Map<String, String> attributes) {341            Set<String> tags = new HashSet<>();342            String cat = attributes.getOrDefault("category", "");343            if (!cat.isBlank()) tags.add("cat:" + cat);344            if (score > 0.8) tags.add("top_score");345            return tags;346        }347    }348349    // ─────────────────────────────────────────────350    //  端到端 Demo351    // ─────────────────────────────────────────────352353    public static Map<String, Object> runDemo() {354        DualBufferIndex forwardIndex = new DualBufferIndex();355        OrderedInvertedIndex invertedIndex = new OrderedInvertedIndex();356        RealtimeIndexUpdater updater = new RealtimeIndexUpdater(forwardIndex, invertedIndex);357358        // 1. 全量加载359        List<ItemFeature> bulk = new ArrayList<>();360        for (int i = 1; i <= 100; i++) {361            bulk.add(new ItemFeature((long) i, i * 0.01,362                    Map.of("category", i % 2 == 0 ? "game" : "food",363                           "author_id", String.valueOf(i % 10))));364        }365        forwardIndex.reload(bulk);366367        // 2. 实时更新：模拟 Kafka 触发消息368        updater.consume(List.of(369                new IndexUpdateTrigger(1L, "CONTENT_UPDATED", System.currentTimeMillis()),370                new IndexUpdateTrigger(2L, "INTERACTION_CHANGED", System.currentTimeMillis()),371                new IndexUpdateTrigger(3L, "CONTENT_UPDATED", System.currentTimeMillis())372        ));373        // 模拟下线触发（status=offline 的物品会走删除路径）374        updater.consume(List.of(375                new IndexUpdateTrigger(99L, "OFFLINE", System.currentTimeMillis())376        ));377378        // 3. 倒排查询379        List<InvertedEntry> foodItems = invertedIndex.query("cat:food", 5);380        List<InvertedEntry> topItems = invertedIndex.query("top_score", 10);381382        System.out.println("Food category top 5: " + foodItems.stream()383                .map(e -> e.itemId + "(" + String.format("%.2f", e.sortScore) + ")")384                .toList());385        System.out.println("Top score items: " + topItems.stream()386                .map(e -> e.itemId + "(" + String.format("%.2f", e.sortScore) + ")")387                .toList());388389390        return Map.of(391                "forward_index_size", forwardIndex.size(),392                "food_tag_size", invertedIndex.tagSize("cat:food")393        );394    }395396    public static void main(String[] args) {397        System.out.println("=== ch06 物品索引服务 Demo ===");398        Map<String, Object> result = runDemo();399        System.out.println("Result: " + result);400    }401}