1. 题目背景解析

这道来自虾皮（Shopee）2026年校园招聘的笔试真题，反映了当前互联网企业对候选人算法能力的核心考察方向。作为东南亚领先的电商平台，虾皮的笔试题目往往紧密结合实际业务场景，这道3月10日的题目也不例外。

从行业趋势来看，2026年的电商平台技术面试呈现三个明显特征：

1. 算法题更注重时间/空间复杂度的双重优化
2. 题目描述会模拟真实业务场景
3. 对边界条件的考察更加严格

2. 题目还原与题意分析

根据多位参与者的回忆，这道题的完整描述大致如下：

给定一个商品ID列表和对应的访问次数，要求实现两个功能：

1. 实时统计最近N分钟内访问量最高的K个商品
2. 支持动态调整统计时间窗口N的值

输入格式示例：

[ {"item_id": "A123", "timestamp": 1710000000, "count": 5}, {"item_id": "B456", "timestamp": 1710000060, "count": 3}, ... ]

2.1 核心考点拆解

这道题主要考察以下技术点：

1. 滑动窗口算法的实现
2. 优先级队列（堆）的应用
3. 时间序列数据的处理
4. 系统设计中的trade-off思维

3. 解决方案设计

3.1 基础解法：滑动窗口+最大堆

最直观的解法是维护一个滑动窗口，配合最大堆实现TopK查询：

from collections import defaultdict import heapq class HotItemsTracker: def __init__(self, N): self.window_size = N * 60 # 转换为秒 self.item_counts = defaultdict(int) self.max_heap = [] self.data = [] def add_record(self, record): current_time = record['timestamp'] # 移除过期数据 while self.data and self.data[0]['timestamp'] < current_time - self.window_size: expired = self.data.pop(0) self.item_counts[expired['item_id']] -= expired['count'] # 添加新数据 self.data.append(record) self.item_counts[record['item_id']] += record['count'] # 更新堆 if record['item_id'] in [x[1] for x in self.max_heap]: self.max_heap = [] for item, count in self.item_counts.items(): if count > 0: heapq.heappush(self.max_heap, (-count, item)) else: heapq.heappush(self.max_heap, (-record['count'], record['item_id'])) def get_top_k(self, k): return [heapq.heappop(self.max_heap)[1] for _ in range(min(k, len(self.max_heap)))]

3.2 复杂度分析

• 时间复杂度：

  • 添加记录：O(M) 最坏情况（M为窗口内记录数）
  • 查询TopK：O(K log N)

• 空间复杂度：O(N) 存储窗口内所有记录

4. 优化方案探讨

4.1 延迟删除优化

基础解法中每次添加记录都要检查过期数据，可以改为惰性删除：

def add_record_optimized(self, record): current_time = record['timestamp'] # 只在堆顶元素过期时清理 while self.max_heap and self.item_counts[self.max_heap[0][1]] == 0: heapq.heappop(self.max_heap) # 添加新记录 self.item_counts[record['item_id']] += record['count'] heapq.heappush(self.max_heap, (-self.item_counts[record['item_id']], record['item_id']))

4.2 时间分片优化

对于大规模数据，可以采用时间分片策略：

1. 将时间窗口分为多个桶（如每分钟一个桶）
2. 维护每个桶的独立计数
3. 查询时合并活跃桶的数据

5. 生产环境考量

在实际业务场景中，还需要考虑：

1. 分布式扩展：当数据量极大时，如何分片处理
2. 容错机制：如何处理节点故障
3. 数据一致性：在分布式环境下保证统计准确性

6. 面试技巧

遇到这类题目时建议：

1. 先明确需求边界（时间精度、数据规模等）
2. 从暴力解法开始，逐步优化
3. 主动讨论不同方案的trade-off
4. 考虑异常情况（如时间戳乱序）

7. 变种题目预测

类似的变种题目可能包括：

1. 加入商品类目维度统计
2. 支持多级时间窗口（最近1分钟/1小时/1天）
3. 处理用户行为序列而非简单计数

8. 代码测试用例

完整的测试应该包含：

def test_hot_items(): tracker = HotItemsTracker(N=5) # 5分钟窗口 test_data = [ {"item_id": "A", "timestamp": 1000, "count": 2}, {"item_id": "B", "timestamp": 1005, "count": 3}, {"item_id": "A", "timestamp": 1010, "count": 1}, {"item_id": "C", "timestamp": 1100, "count": 5} # 超出窗口 ] for record in test_data: tracker.add_record(record) assert tracker.get_top_k(2) == ["A", "B"]

9. 性能优化实验

我们对不同数据规模下的性能进行了测试：

10. 工程实践建议

在实际项目中实现类似功能时：

1. 使用Redis的Sorted Set可以简化开发
2. 考虑使用Apache Flink等流处理框架
3. 对于精确度要求不高的场景，可以采样处理

这种题型在各大厂的笔试中越来越常见，掌握其核心解题模式对求职者至关重要。关键是要理解滑动窗口与堆结构的配合使用，以及如何在工程实践中进行合理优化。