合并策略设计：时间窗口聚合、批量请求封装、优先级队列与去重

合并策略是请求合并引擎的核心，它决定了如何将多个独立请求整合为一批，在功耗与延迟之间取得最优平衡。本章详细讲解四种关键策略的设计原理、实现流程与调试方法：时间窗口聚合、批量请求封装、优先级队列与去重。这四种策略通常串行或并行组合形成一个完整的合并管道。

3.1 时间窗口聚合

时间窗口聚合是最基础的合并策略：将到达的请求在固定或动态的时间窗内暂存，窗口关闭时一次性合并该窗口内的所有请求。核心参数是窗口长度 W（毫秒），它直接决定延迟与合并收益的平衡点。

3.1.1 固定窗口实现

class TimeWindowAggregator:
    def __init__(self, window_ms=100, batch_handler=None):
        self.window_ms = window_ms
        self.batch = []
        self.timer = None
        self.handler = batch_handler

    def add_request(self, req):
        self.batch.append(req)
        if self.timer is None:
            self.timer = threading.Timer(self.window_ms / 1000.0, self._flush)
            self.timer.start()

    def _flush(self):
        if self.batch:
            self.handler(self.batch)
            self.batch = []
        self.timer = None

3.1.2 调试与调优

3.2 批量请求封装

当多个请求被聚合后，需要将它们封装成单个网络请求体。常见的封装格式包括JSON数组、MessagePack序列化或自定义二进制协议。封装逻辑必须考虑请求间的隔离性、错误独立性和幂等性。

3.2.1 JSON数组封装示例

def batch_encode(requests: list[dict]) -> bytes:
    payload = {"batch": requests, "count": len(requests)}
    return json.dumps(payload, ensure_ascii=False).encode("utf-8")

3.2.2 批量响应拆分

服务端返回的批量响应必须按原请求顺序拆分，客户端通过序号映射或请求ID还原每个独立结果。

def batch_decode(response: bytes) -> list[dict]:
    data = json.loads(response)
    return data["results"]  # 预期是一个与请求顺序一致的数组

3.3 优先级队列

当合并窗口内有多个紧急请求（如用户交互触发的请求）和非紧急请求（如后台同步）时，优先级队列确保高优请求尽早被处理和发送。优先级通常以整型值表示（0最高优先级）。

3.3.1 带优先级的合并流程

1. 每个请求携带priority字段（0–7），默认优先级=4。
2. 时间窗口内暂存请求时，按优先级插入最小堆或使用heapq。
3. 窗口flush时，弹出所有请求并按优先级降序排序，保证高优先请求在批量的前面（某些服务顺序敏感）。
4. 可选：设置优先窗口——高优先级请求可触发一种短窗口，快速发送批。

3.3.2 Python优先级队列合并器

import heapq
import threading

class PriorityBatchQueue:
    def __init__(self, window_ms=100):
        self._heap = []
        self._lock = threading.Lock()
        self.window_ms = window_ms
        self._timer = None

    def add(self, req):
        priority = req.get("priority", 4)
        with self._lock:
            heapq.heappush(self._heap, (-priority, req))  # 负数实现最大堆
        if self._timer is None:
            self._timer = threading.Timer(self.window_ms/1000, self.flush)
            self._timer.start()

    def flush(self):
        with self._lock:
            items = [heapq.heappop(self._heap)[1] for _ in range(len(self._heap))]
        if items:
            items_sorted = list(reversed(items))  # 高优在前
            send_batch(items_sorted)
        self._timer = None

3.4 去重

同一用户或同一设备在短时间内可能发出相同请求（如按钮重复点击），直接合并会导致重复处理。去重策略基于请求的唯一键（通常由请求URL + 参数哈希或客户端生成的ID组成）去除重复项。

3.4.1 去重实现步骤

• 生成请求指纹：对URL + 请求体 + 时间戳前缀（如5秒粒度）做SHA256。
• 维护滑动窗口集合：使用collections.deque或redis zset存储最近N秒的指纹。
• 判断重复：新请求指纹若已存在，则丢弃；否则记录并加入聚合。

3.4.2 内存去重示例

class DeduplicationFilter:
    def __init__(self, window_sec=5):
        self.window_sec = window_sec
        self._seen = set()
        self._expire_timer = threading.Timer(window_sec, self._cleanup)
        self._expire_timer.start()

    def is_duplicate(self, fingerprint: str) -> bool:
        if fingerprint in self._seen:
            return True
        self._seen.add(fingerprint)
        return False

    def _cleanup(self):
        self._seen.clear()
        self._expire_timer = threading.Timer(self.window_sec, self._cleanup)
        self._expire_timer.start()

3.5 四种策略的性能对比

3.6 调试与部署要点

• 日志关键指标：每次flush记录batch_size、窗口实际时长、去重丢弃数、优先级倒挂次数。
• 负载测试：模拟每秒1000个请求，观察窗口是否按时flush、是否有请求被遗忘。
• CPU profile：去重哈希计算和优先级堆操作为热点，考虑使用xxhash替代SHA256。
• 网络中断容错：批量封装后发送失败应回退到逐请求重试，避免整批丢失。