import threading
import time
from abc import ABC, abstractmethod
from collections import deque
from typing import Generic, TypeVar
T = TypeVar('T')
class TimeWindow(Generic[T]):
def __init__(self, start_time: int, value: T):
self.start_time: int = start_time
self.value: T = value
class SlidingWindow(ABC, Generic[T]):
"""
通用滑动窗口抽象基类
仅负责时间分片骨架（桶对齐 / 桶复用 / 并发安全 / 桶生命周期），
不感知任何业务字段。具体业务桶类型由子类通过泛型参数 T 指定，
具体业务读写接口由子类实现。
通过 total_window_ms（总窗口时长）和 window_duration_ms（每个小窗口时长）控制
窗口数量 = total_window_ms / window_duration_ms。
构造时预分配 window_count 个桶，运行期通过环形 deque 原地复用，不再创建或丢弃桶对象。
"""
def __init__(self, total_window_ms: int, window_duration_ms: int):
if total_window_ms <= 0 or window_duration_ms <= 0:
raise ValueError("total_window_ms and window_duration_ms must be positive")
if total_window_ms % window_duration_ms != 0:
raise ValueError("total_window_ms must be divisible by window_duration_ms")
self._total_window_ms = total_window_ms
self._window_duration_ms = window_duration_ms
self._window_count = total_window_ms // window_duration_ms
self._buckets: deque[TimeWindow[T]] = deque(maxlen=self._window_count)
self._lock = threading.Lock()
for _ in range(self._window_count):
self._buckets.append(TimeWindow(-1, self._create_bucket()))
@abstractmethod
def _create_bucket(self) -> T:
"""
创建一个新的桶值实例，由子类决定具体桶类型
仅在 __init__ 预分配阶段调用，运行期不再调用
:return: 新桶值实例
"""
@abstractmethod
def _reset_bucket(self, bucket: T) -> None:
"""
把已存在的桶值复位到与新建状态等价
子类需要把所有业务字段恢复初始值
:param bucket: 待复位的桶值
:return:
"""
def _now_ms(self) -> int:
return int(time.perf_counter() * 1000)
def current_bucket(self) -> TimeWindow[T]:
"""
根据当前时间戳获取当前窗口， 同时清理已过期的时间窗口, 保证对象复用防止重复创建对象
通过 time_id 对窗口数量取模定位环形数组中的桶位置，
若桶的起始时间与当前时间窗口不匹配则复位后复用
:return: 当前时间窗口对应的桶
"""
now_ms = self._now_ms()
window_start = now_ms - (now_ms % self._window_duration_ms)
tail = self._buckets[-1]
if tail.start_time == window_start:
return tail
else:
with self._lock:
tail = self._buckets[-1]
if tail.start_time == window_start:
return tail
oldest = self._buckets.popleft()
self._reset_bucket(oldest.value)
oldest.start_time = window_start
self._buckets.append(oldest)
return oldest
def get_all_buckets(self) -> list[T]:
"""
无锁返回当前 deque 中所有【有效】桶的快照列表
过滤掉哨兵桶（window_start=-1）与已过期桶
利用 CPython deque 的线程安全性：list(deque) 不会因并发修改抛 RuntimeError
:return: 有效桶引用列表
"""
now_ms = self._now_ms()
cutoff = now_ms - self._total_window_ms
return [b.value for b in self._buckets if b.start_time > cutoff]