vincent/EnterpriseArchitect
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(BIZ-26): 实现 API 请求优先级队列 + 令牌桶限流器

Browse Source 
- 新增 scripts/rate_limiter.py 核心模块
  - TokenBucket: 令牌桶限流器（40 RPM 上限）
  - RequestScheduler: 四级优先级请求队列调度器
  - CacheManager: 查询结果缓存（分 TTL 策略）
  - retry_with_backoff: 指数退避重试装饰器
  - CoordinatedPoller: COO 统一轮询器

- 新增 scripts/test_rate_limiter.py 测试套件
  - 覆盖令牌桶、缓存、队列、重试、轮询、压力测试
  - 所有测试通过 ✅

- 新增 docs/BIZ-26-限流器使用文档.md
  - API 参考、使用示例、集成指南
  - 缓存策略、降级策略、监控调试

实现参考：plans/BIZ-13_运行稳定性保障方案.md

Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
bizwings
2026-06-23 07:09:39 +08:00
parent 80ef3c2796
commit 7f1edfb2fd
3 changed files with 1368 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/BIZ-26-限流器使用文档.md
+377
View File
@@ -0,0 +1,377 @@
# BIZ-26 限流器使用文档
> 模块：`scripts/rate_limiter.py`
> 测试：`scripts/test_rate_limiter.py`
> 实现日期：2026-06-23
> 作者：徐聪（costcodev）
---
## 一、功能概述
本模块实现了 BIZ-13 运行稳定性保障方案中的 API 限流优化功能：
1. **令牌桶限流器**：40 RPM 上限，防止触发 API 429 错误
2. **四级优先级队列**：紧急 > 高 > 正常 > 低
3. **智能降级策略**：高优先级等待，低优先级切备用模型
4. **缓存管理器**：按数据类型设置不同 TTL
5. **COO 统一轮询**：减少重复请求
6. **指数退避重试**：自动处理临时失败
---
## 二、快速开始
### 2.1 基本用法
```python
from scripts.rate_limiter import RequestScheduler, Priority
# 创建调度器（40 RPM）
scheduler = RequestScheduler(rate=40/60, capacity=40)
scheduler.start()
# 提交请求
def my_callback(data):
# 实际 API 调用逻辑
return process_data(data)
request_id = scheduler.submit(
payload={"task": "process_workboard"},
priority=Priority.NORMAL,
callback=my_callback
)
# 等待完成后关闭
time.sleep(5)
scheduler.stop()
```
### 2.2 优先级示例
```python
# 紧急任务（Vincent 直接下达）
scheduler.submit(payload=data, priority=Priority.URGENT, callback=handler)
# 阻塞性任务（依赖下游完成）
scheduler.submit(payload=data, priority=Priority.HIGH, callback=handler)
# 常规任务
scheduler.submit(payload=data, priority=Priority.NORMAL, callback=handler)
# 后台优化任务
scheduler.submit(payload=data, priority=Priority.LOW, callback=handler)
```
### 2.3 缓存使用
```python
from scripts.rate_limiter import CacheManager
cache = CacheManager()
# 缓存 WorkBoard 结果（TTL 5 分钟）
cache.set("workboard", "todo_list", result_data)
# 读取缓存
cached = cache.get("workboard", "todo_list")
if cached is None:
# 缓存未命中，重新查询
result = query_workboard()
cache.set("workboard", "todo_list", result)
# 查看缓存统计
stats = cache.get_stats()
print(f"缓存条目：{stats['total_entries']}")
```
---
## 三、API 参考
### 3.1 TokenBucket（令牌桶）
```python
bucket = TokenBucket(rate=40/60, capacity=40)
# 尝试消费令牌（立即返回）
if bucket.consume():
send_request()
else:
# 令牌不足，等待或降级
pass
# 等待令牌（阻塞直到获取或超时）
got_token = bucket.wait_for_token(timeout=5.0)
# 查看状态
status = bucket.get_status()
# 返回：{"tokens": 35.5, "capacity": 40, "rate_per_minute": 40.0, ...}
```
### 3.2 RequestScheduler（请求调度器）
```python
scheduler = RequestScheduler(
rate=40/60, # 令牌生成速率（个/秒）
capacity=40, # 桶容量
enable_cache=True # 启用缓存
)
# 启动工作线程
scheduler.start()
# 提交异步请求
request_id = scheduler.submit(
payload={"task": "data"},
priority=Priority.NORMAL,
callback=my_handler,
fallback_model="deepseek-v4-pro"
)
# 提交同步请求（阻塞直到完成）
result = scheduler.submit_sync(
payload={"task": "data"},
priority=Priority.URGENT,
timeout=10.0
)
# 查看状态
status = scheduler.get_status()
# 停止调度器
scheduler.stop()
```
### 3.3 CacheManager（缓存管理器）
```python
cache = CacheManager()
# 设置缓存（自动 TTL）
cache.set("workboard", query_key, value) # 5 分钟
cache.set("config", "agent_list", agents) # 1 小时
cache.set("knowledge", "api_docs", docs) # 1 天
# 自定义 TTL
cache.set("custom", key, value, ttl=600) # 10 分钟
# 读取缓存
value = cache.get("workboard", query_key)
# 删除缓存
cache.delete("workboard", query_key)
# 清理过期缓存
cleaned = cache.clear_expired()
# 查看统计
stats = cache.get_stats()
```
### 3.4 retry_with_backoff（重试装饰器）
```python
from rate_limiter import retry_with_backoff
@retry_with_backoff(
max_retries=3, # 最多重试 3 次
base_delay=1.0, # 基础延迟 1 秒
exponential_base=2, # 指数底数
jitter=True, # 添加随机抖动
exceptions=(RateLimitError, NetworkError)
)
def call_api():
return requests.get(url)
```
### 3.5 CoordinatedPoller（统一轮询器）
```python
from rate_limiter import CoordinatedPoller
# 创建轮询器（15 分钟轮询一次）
poller = CoordinatedPoller(scheduler, poll_interval=15*60)
# 订阅轮询结果
def on_new_data(result):
broadcast_to_agents(result)
poller.subscribe(on_new_data)
# 启动轮询
poller.start()
# 停止轮询
poller.stop()
```
---
## 四、缓存策略
| 数据类型 | TTL | 说明 |
|----------|-----|------|
| `workboard` | 5 分钟 | WorkBoard 卡片状态，高频变化 |
| `config` | 1 小时 | Agent 配置、技能列表，低频变化 |
| `knowledge` | 1 天 | 知识库内容，基本不变 |
| `user` | 1 天 | 用户信息、权限配置 |
---
## 五、降级策略
### 5.1 令牌不足时的处理
| 优先级 | 策略 |
|--------|------|
| URGENT (1) | 无限等待，直到获取令牌 |
| HIGH (2) | 无限等待，直到获取令牌 |
| NORMAL (3) | 等待 2 秒，失败则放回队列稍后重试 |
| LOW (4) | 等待 2 秒，失败则丢弃或切换到备用模型 |
### 5.2 模型降级链
```
主模型 (qwen3.5-397b)
↓ RPM 不足
备用模型 (deepseek-v4-pro)
↓ RPM 不足
本地模型 或 等待
```
---
## 六、监控与调试
### 6.1 查看调度器状态
```python
status = scheduler.get_status()
print(f"队列大小：{status['queue_size']}")
print(f"令牌数：{status['token_bucket']['tokens']}")
print(f"已完成：{status['stats']['completed_requests']}")
print(f"失败：{status['stats']['failed_requests']}")
print(f"降级：{status['stats']['fallback_requests']}")
```
### 6.2 查看缓存统计
```python
stats = cache.get_stats()
print(f"总条目：{stats['total_entries']}")
print(f"有效条目：{stats['valid_entries']}")
print(f"过期条目：{stats['expired_entries']}")
print(f"按类别：{stats['by_category']}")
```
---
## 七、测试
运行测试套件：
```bash
cd /home/vincent/.openclaw/workspace/costcodev/EnterpriseArchitect
python3 scripts/test_rate_limiter.py
```
测试覆盖：
- ✅ 令牌桶限流
- ✅ 缓存管理
- ✅ 优先级队列
- ✅ 重试装饰器
- ✅ 统一轮询器
- ✅ 压力测试（50 请求）
---
## 八、集成示例
### 8.1 与 Multica CLI 集成
```python
import subprocess
import json
from rate_limiter import RequestScheduler, Priority, CacheManager
scheduler = RequestScheduler(rate=40/60, capacity=40)
cache = CacheManager()
scheduler.start()
def query_workboard():
"""查询 WorkBoard（带缓存）"""
# 先查缓存
cached = cache.get("workboard", "all_cards")
if cached:
return cached
# 缓存未命中，调用 CLI
result = subprocess.run(
["multica", "workboard", "list", "--json"],
capture_output=True,
text=True
)
data = json.loads(result.stdout)
# 更新缓存
cache.set("workboard", "all_cards", data)
return data
# 提交查询请求
request_id = scheduler.submit(
payload="query_workboard",
priority=Priority.NORMAL,
callback=lambda _: query_workboard()
)
```
### 8.2 Agent 心跳集成
```python
# 在 Heartbeat 中统一使用限流器
def heartbeat_check():
# 通过调度器提交所有检查任务
scheduler.submit(
payload="check_workboard",
priority=Priority.HIGH,
callback=check_workboard
)
scheduler.submit(
payload="check_multica",
priority=Priority.HIGH,
callback=check_multica_issues
)
scheduler.submit(
payload="update_memory",
priority=Priority.LOW,
callback=update_memory_log
)
```
---
## 九、注意事项
1. **令牌速率配置**：根据实际 API 限制调整 `rate` 参数
2. **缓存 TTL**：根据数据变化频率调整，避免过期数据
3. **工作线程**：记得调用 `start()` 和 `stop()` 管理生命周期
4. **异常处理**：回调函数中的异常会被捕获并记录，不会中断工作线程
5. **线程安全**：所有组件都是线程安全的，可在多线程环境使用
---
## 十、TODO
- [ ] 接入实际的 Multica CLI 调用
- [ ] 添加 Prometheus 监控指标导出
- [ ] 支持动态调整限流参数
- [ ] 添加请求日志持久化
- [ ] 支持多个模型池的自动切换
---
> 文档版本：v1.0
> 最后更新：2026-06-23
> 维护者：徐聪（costcodev）
scripts/rate_limiter.py
+685
View File
@@ -0,0 +1,685 @@
"""
BIZ-26: API 请求优先级队列 + 令牌桶限流器
实现方案参考：plans/BIZ-13_运行稳定性保障方案.md
功能清单：
1. 四级优先级请求队列（紧急 > 高 > 正常 > 低）
2. 令牌桶限流器（40 RPM 上限）
3. 超限自动降级和等待策略
4. 请求合并（COO 统一轮询）
5. 查询结果缓存（WorkBoard 5 分钟、配置 1 小时、知识库 1 天）
作者：徐聪（costcodev）
日期：2026-06-23
"""
import time
import threading
import queue
import hashlib
import json
from typing import Any, Callable, Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import IntEnum
from datetime import datetime, timedelta
# ============================================================================
# 优先级枚举
# ============================================================================
class Priority(IntEnum):
"""请求优先级：数值越小优先级越高"""
URGENT = 1 # 紧急：Vincent 直接任务
HIGH = 2 # 高：阻塞性任务
NORMAL = 3 # 正常：常规任务
LOW = 4 # 低：后台优化任务
# ============================================================================
# 请求数据类
# ============================================================================
@dataclass(order=True)
class Request:
"""优先级队列中的请求项"""
priority: int
timestamp: float = field(compare=False)
request_id: str = field(compare=False)
payload: Any = field(compare=False)
callback: Optional[Callable] = field(compare=False, default=None)
fallback_model: Optional[str] = field(compare=False, default=None)
def __post_init__(self):
if self.timestamp is None:
self.timestamp = time.time()
if self.request_id is None:
self.request_id = self._generate_id()
@staticmethod
def _generate_id() -> str:
"""生成请求 ID"""
return hashlib.md5(f"{time.time()}-{threading.current_thread().ident}".encode()).hexdigest()[:12]
# ============================================================================
# 令牌桶限流器
# ============================================================================
class TokenBucket:
"""
令牌桶限流器
参数：
rate: 令牌生成速率（个/秒），默认 40 RPM = 0.67 个/秒
capacity: 桶容量（最大令牌数），默认 40
"""
def __init__(self, rate: float = 40/60, capacity: int = 40):
self.rate = rate # 令牌/秒
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self._lock = threading.Lock()
def _refill(self) -> None:
"""补充令牌（内部调用，需要持有锁）"""
now = time.time()
elapsed = now - self.last_update
new_tokens = elapsed * self.rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_update = now
def consume(self, tokens: int = 1) -> bool:
"""
尝试消费令牌
返回：
True: 成功消费
False: 令牌不足
"""
with self._lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def wait_for_token(self, timeout: Optional[float] = None) -> bool:
"""
等待直到有可用令牌
参数：
timeout: 最大等待时间（秒），None 表示无限等待
返回：
True: 成功获取令牌
False: 超时
"""
start_time = time.time()
while True:
if self.consume():
return True
if timeout is not None:
elapsed = time.time() - start_time
if elapsed >= timeout:
return False
# 计算等待时间（直到下一个令牌生成）
with self._lock:
self._refill()
if self.tokens < 1:
wait_time = (1 - self.tokens) / self.rate
else:
wait_time = 0.01
# 等待后重试
time_to_wait = min(wait_time, 0.1) # 最多等待 100ms
if timeout is not None:
remaining = timeout - (time.time() - start_time)
if remaining <= 0:
return False
time_to_wait = min(time_to_wait, remaining)
time.sleep(time_to_wait)
def get_status(self) -> Dict[str, Any]:
"""获取限流器状态"""
with self._lock:
self._refill()
return {
"tokens": round(self.tokens, 2),
"capacity": self.capacity,
"rate_per_second": round(self.rate, 3),
"rate_per_minute": round(self.rate * 60, 1),
"utilization": round(1 - self.tokens / self.capacity, 2)
}
# ============================================================================
# 缓存管理器
# ============================================================================
@dataclass
class CacheEntry:
"""缓存条目"""
value: Any
expires_at: float
created_at: float = field(default_factory=time.time)
access_count: int = field(default=0)
class CacheManager:
"""
查询结果缓存管理器
缓存策略：
- WorkBoard 状态：5 分钟
- Agent 配置：1 小时
- 知识库内容：1 天
- 用户信息：1 天
"""
# 默认 TTL 配置（秒）
DEFAULT_TTL = {
"workboard": 5 * 60, # 5 分钟
"config": 1 * 60 * 60, # 1 小时
"knowledge": 24 * 60 * 60, # 1 天
"user": 24 * 60 * 60, # 1 天
}
def __init__(self):
self._cache: Dict[str, CacheEntry] = {}
self._lock = threading.Lock()
def _generate_key(self, category: str, query: Any) -> str:
"""生成缓存键"""
query_str = json.dumps(query, sort_keys=True) if not isinstance(query, str) else query
return hashlib.md5(f"{category}:{query_str}".encode()).hexdigest()
def get(self, category: str, query: Any) -> Optional[Any]:
"""
获取缓存
参数：
category: 缓存类别（workboard/config/knowledge/user）
query: 查询条件（用于生成缓存键）
返回：
缓存值，如果不存在或已过期则返回 None
"""
key = self._generate_key(category, query)
with self._lock:
entry = self._cache.get(key)
if entry is None:
return None
# 检查是否过期
if time.time() > entry.expires_at:
del self._cache[key]
return None
# 更新访问计数
entry.access_count += 1
return entry.value
def set(self, category: str, query: Any, value: Any, ttl: Optional[int] = None) -> None:
"""
设置缓存
参数：
category: 缓存类别
query: 查询条件
value: 缓存值
ttl: 存活时间（秒），None 表示使用默认值
"""
key = self._generate_key(category, query)
if ttl is None:
ttl = self.DEFAULT_TTL.get(category, 300) # 默认 5 分钟
with self._lock:
self._cache[key] = CacheEntry(
value=value,
expires_at=time.time() + ttl
)
def delete(self, category: str, query: Any) -> bool:
"""删除缓存"""
key = self._generate_key(category, query)
with self._lock:
if key in self._cache:
del self._cache[key]
return True
return False
def clear_expired(self) -> int:
"""清理所有过期缓存，返回清理数量"""
now = time.time()
with self._lock:
expired_keys = [k for k, v in self._cache.items() if now > v.expires_at]
for key in expired_keys:
del self._cache[key]
return len(expired_keys)
def get_stats(self) -> Dict[str, Any]:
"""获取缓存统计"""
now = time.time()
with self._lock:
total = len(self._cache)
expired = sum(1 for v in self._cache.values() if now > v.expires_at)
# 按类别统计
by_category: Dict[str, int] = {}
for key, entry in self._cache.items():
# 从 key 中提取 category（格式：category:hash）
category = key.split(":")[0] if ":" in key else "unknown"
by_category[category] = by_category.get(category, 0) + 1
return {
"total_entries": total,
"expired_entries": expired,
"valid_entries": total - expired,
"by_category": by_category
}
def clear(self) -> None:
"""清空所有缓存"""
with self._lock:
self._cache.clear()
# ============================================================================
# 请求调度器
# ============================================================================
class RequestScheduler:
"""
请求调度器：结合优先级队列和令牌桶限流
功能：
1. 接收不同优先级的请求
2. 按优先级和 FIF0 顺序调度
3. 通过令牌桶控制发送速率
4. 支持降级策略（低优先级切备用模型）
"""
def __init__(
self,
rate: float = 40/60,
capacity: int = 40,
enable_cache: bool = True
):
self.token_bucket = TokenBucket(rate=rate, capacity=capacity)
self.cache = CacheManager() if enable_cache else None
# 优先级队列（使用 heap 实现）
self.request_queue: queue.PriorityQueue[Request] = queue.PriorityQueue()
# 工作线程
self._worker_thread: Optional[threading.Thread] = None
self._running = False
self._lock = threading.Lock()
# 统计信息
self.stats = {
"total_requests": 0,
"completed_requests": 0,
"failed_requests": 0,
"fallback_requests": 0,
"cache_hits": 0,
"cache_misses": 0,
}
def start(self) -> None:
"""启动调度器工作线程"""
if self._running:
return
self._running = True
self._worker_thread = threading.Thread(target=self._worker_loop, daemon=True)
self._worker_thread.start()
def stop(self) -> None:
"""停止调度器"""
self._running = False
if self._worker_thread:
self._worker_thread.join(timeout=5.0)
def _worker_loop(self) -> None:
"""工作线程主循环"""
while self._running:
try:
# 从队列获取请求（带超时）
request = self.request_queue.get(timeout=1.0)
self._process_request(request)
except queue.Empty:
continue
except Exception as e:
# 记录错误但不中断工作线程
print(f"[RequestScheduler] Worker error: {e}")
def _process_request(self, request: Request) -> None:
"""
处理单个请求
策略：
1. 高优先级（URGENT/HIGH）：等待令牌
2. 低优先级（NORMAL/LOW）：尝试获取令牌，失败则降级或丢弃
"""
self.stats["total_requests"] += 1
# 尝试获取令牌
if request.priority <= Priority.HIGH:
# 高优先级：无限等待
got_token = self.token_bucket.wait_for_token(timeout=None)
else:
# 低优先级：最多等待 2 秒
got_token = self.token_bucket.wait_for_token(timeout=2.0)
if got_token:
# 成功获取令牌，执行请求
self._execute_request(request)
else:
# 未能获取令牌，执行降级策略
self._handle_fallback(request)
def _execute_request(self, request: Request) -> None:
"""执行请求"""
try:
if request.callback:
result = request.callback(request.payload)
self.stats["completed_requests"] += 1
return result
else:
self.stats["completed_requests"] += 1
except Exception as e:
self.stats["failed_requests"] += 1
print(f"[RequestScheduler] Request {request.request_id} failed: {e}")
raise
def _handle_fallback(self, request: Request) -> None:
"""处理降级（令牌不足）"""
self.stats["fallback_requests"] += 1
if request.priority == Priority.LOW:
# 低优先级：直接丢弃或切换到备用模型
print(f"[RequestScheduler] Low priority request {request.request_id} dropped due to rate limit")
else:
# 正常优先级：放回队列稍后重试
request.timestamp = time.time()
self.request_queue.put(request)
def submit(
self,
payload: Any,
priority: Priority = Priority.NORMAL,
callback: Optional[Callable] = None,
fallback_model: Optional[str] = None,
request_id: Optional[str] = None
) -> str:
"""
提交请求到调度队列
参数：
payload: 请求数据
priority: 优先级
callback: 回调函数
fallback_model: 备用模型名称
request_id: 请求 ID（可选，默认自动生成）
返回：
请求 ID
"""
req = Request(
priority=priority,
timestamp=time.time(),
request_id=request_id,
payload=payload,
callback=callback,
fallback_model=fallback_model
)
self.request_queue.put(req)
return req.request_id
def submit_sync(
self,
payload: Any,
priority: Priority = Priority.NORMAL,
timeout: Optional[float] = None
) -> Any:
"""
同步提交并等待结果
参数：
payload: 请求数据
priority: 优先级
timeout: 超时时间（秒）
返回：
请求结果
"""
result_holder = {"result": None, "error": None, "done": False}
condition = threading.Condition()
def callback(data):
with condition:
try:
# 实际执行逻辑（这里只是一个占位符）
result_holder["result"] = data
except Exception as e:
result_holder["error"] = e
finally:
result_holder["done"] = True
condition.notify_all()
# 提交请求
self.submit(payload=payload, priority=priority, callback=lambda _: callback(payload))
# 等待结果
with condition:
if not result_holder["done"]:
condition.wait(timeout=timeout)
if result_holder["error"]:
raise result_holder["error"]
return result_holder["result"]
def get_queue_size(self) -> int:
"""获取当前队列大小"""
return self.request_queue.qsize()
def get_status(self) -> Dict[str, Any]:
"""获取调度器状态"""
return {
"running": self._running,
"queue_size": self.get_queue_size(),
"token_bucket": self.token_bucket.get_status(),
"cache": self.cache.get_stats() if self.cache else None,
"stats": self.stats.copy()
}
# ============================================================================
# 重试装饰器
# ============================================================================
def retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
exponential_base: int = 2,
jitter: bool = True,
exceptions: Tuple = (Exception,)
):
"""
指数退避重试装饰器
参数：
max_retries: 最大重试次数
base_delay: 基础延迟（秒）
exponential_base: 指数底数
jitter: 是否添加随机抖动
exceptions: 需要重试的异常类型
"""
import random
def decorator(func: Callable) -> Callable:
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt == max_retries:
break
# 计算延迟时间
delay = base_delay * (exponential_base ** attempt)
if jitter:
delay += random.uniform(0, base_delay)
print(f"[retry_with_backoff] Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s...")
time.sleep(delay)
raise last_exception
return wrapper
return decorator
# ============================================================================
# COO 统一轮询器（请求合并）
# ============================================================================
class CoordinatedPoller:
"""
COO 统一轮询器：替代各 Agent 独立轮询
功能：
1. 定期轮询 WorkBoard
2. 广播结果给所有订阅者
3. 减少总请求数（40 RPM × N → 40 RPM）
"""
def __init__(self, scheduler: RequestScheduler, poll_interval: int = 15*60):
self.scheduler = scheduler
self.poll_interval = poll_interval # 轮询间隔（秒）
self._subscribers: List[Callable] = []
self._running = False
self._worker: Optional[threading.Thread] = None
def subscribe(self, callback: Callable) -> None:
"""订阅轮询结果"""
self._subscribers.append(callback)
def unsubscribe(self, callback: Callable) -> None:
"""取消订阅"""
if callback in self._subscribers:
self._subscribers.remove(callback)
def start(self) -> None:
"""启动轮询器"""
if self._running:
return
self._running = True
self._worker = threading.Thread(target=self._poll_loop, daemon=True)
self._worker.start()
def stop(self) -> None:
"""停止轮询器"""
self._running = False
if self._worker:
self._worker.join(timeout=5.0)
def _poll_loop(self) -> None:
"""轮询主循环"""
while self._running:
try:
# 执行轮询（这里只是一个框架，实际逻辑需要接入 multica CLI）
result = self._perform_poll()
# 广播给所有订阅者
for subscriber in self._subscribers:
try:
subscriber(result)
except Exception as e:
print(f"[CoordinatedPoller] Subscriber callback error: {e}")
except Exception as e:
print(f"[CoordinatedPoller] Poll error: {e}")
# 等待下一个轮询周期
time.sleep(self.poll_interval)
def _perform_poll(self) -> Dict[str, Any]:
"""
执行实际轮询
TODO: 接入 multica CLI:
- multica issue list --status in_progress
- multica workboard list
"""
# 这里应该调用 multica CLI
# 当前只是返回一个示例结果
return {
"timestamp": datetime.now().isoformat(),
"issues": [],
"workboard_cards": []
}
# ============================================================================
# 使用示例
# ============================================================================
if __name__ == "__main__":
# 创建调度器（40 RPM）
scheduler = RequestScheduler(rate=40/60, capacity=40)
scheduler.start()
# 示例：提交不同优先级的请求
def sample_callback(data):
print(f"Processing: {data}")
time.sleep(0.5) # 模拟处理时间
return "OK"
# 紧急请求
scheduler.submit(
payload={"task": "urgent_task"},
priority=Priority.URGENT,
callback=sample_callback
)
# 正常请求
scheduler.submit(
payload={"task": "normal_task"},
priority=Priority.NORMAL,
callback=sample_callback
)
# 低优先级请求
scheduler.submit(
payload={"task": "low_priority_task"},
priority=Priority.LOW,
callback=sample_callback
)
# 等待处理完成
time.sleep(2)
# 查看状态
print("\n=== Scheduler Status ===")
print(json.dumps(scheduler.get_status(), indent=2))
# 停止调度器
scheduler.stop()
print("\n示例运行完成")
scripts/test_rate_limiter.py
+306
View File
@@ -0,0 +1,306 @@
#!/usr/bin/env python3
"""
BIZ-26 限流器测试脚本
测试场景：
1. 令牌桶限流功能
2. 优先级队列调度
3. 缓存管理器
4. 重试机制
5. 429 错误模拟
运行方式：
python3 scripts/test_rate_limiter.py
"""
import sys
import time
import threading
from datetime import datetime
# 添加脚本目录到路径
sys.path.insert(0, "/home/vincent/.openclaw/workspace/costcodev/EnterpriseArchitect/scripts")
from rate_limiter import (
TokenBucket,
CacheManager,
RequestScheduler,
Priority,
retry_with_backoff,
CoordinatedPoller,
)
def test_token_bucket():
"""测试令牌桶限流器"""
print("=" * 60)
print("测试 1: 令牌桶限流器")
print("=" * 60)
# 创建限流器：40 RPM = 0.67 令牌/秒
bucket = TokenBucket(rate=40/60, capacity=40)
print(f"\n初始状态：{bucket.get_status()}")
# 快速消费 10 个令牌
print("\n快速消费 10 个令牌...")
success_count = 0
for i in range(10):
if bucket.consume():
success_count += 1
print(f"成功消费：{success_count}/10")
print(f"消费后状态：{bucket.get_status()}")
# 测试等待获取令牌
print("\n测试等待获取令牌...")
start = time.time()
got_token = bucket.wait_for_token(timeout=2.0)
elapsed = time.time() - start
print(f"等待耗时：{elapsed:.3f}s, 获取成功：{got_token}")
print(f"等待后状态：{bucket.get_status()}")
print("\n✅ 令牌桶测试完成\n")
def test_cache_manager():
"""测试缓存管理器"""
print("=" * 60)
print("测试 2: 缓存管理器")
print("=" * 60)
cache = CacheManager()
# 测试 WorkBoard 缓存（TTL 5 分钟）
print("\n1. 设置 WorkBoard 缓存（TTL 5 分钟）")
cache.set("workboard", {"query": "status=todo"}, [{"id": "card1", "title": "Test"}])
# 立即读取
result = cache.get("workboard", {"query": "status=todo"})
print(f" 立即读取：{result is not None}")
# 测试配置缓存（TTL 1 小时）
print("\n2. 设置配置缓存（TTL 1 小时）")
cache.set("config", "agent_list", ["costcodev", "secretary", "coo"])
result = cache.get("config", "agent_list")
print(f" 读取配置：{result}")
# 测试缓存统计
print("\n3. 缓存统计")
stats = cache.get_stats()
print(f" 总条目数：{stats['total_entries']}")
print(f" 按类别：{stats['by_category']}")
# 测试缓存删除
print("\n4. 删除缓存")
deleted = cache.delete("workboard", {"query": "status=todo"})
print(f" 删除成功：{deleted}")
result = cache.get("workboard", {"query": "status=todo"})
print(f" 删除后读取：{result is None}")
print("\n✅ 缓存管理器测试完成\n")
def test_priority_queue():
"""测试优先级队列调度"""
print("=" * 60)
print("测试 3: 优先级队列调度（简化版，不启动工作线程）")
print("=" * 60)
scheduler = RequestScheduler(rate=40/60, capacity=40, enable_cache=True)
# 模拟请求处理结果
results = []
def record_result(data):
results.append((time.time(), data))
return data
# 提交不同优先级的请求（不启动工作线程，只测试队列）
print("\n提交请求（按顺序）:")
scheduler.submit(
payload={"task": "normal_1"},
priority=Priority.NORMAL,
callback=record_result
)
print(" 1. 正常优先级：normal_1")
scheduler.submit(
payload={"task": "urgent_1"},
priority=Priority.URGENT,
callback=record_result
)
print(" 2. 紧急优先级：urgent_1")
scheduler.submit(
payload={"task": "low_1"},
priority=Priority.LOW,
callback=record_result
)
print(" 3. 低优先级：low_1")
scheduler.submit(
payload={"task": "high_1"},
priority=Priority.HIGH,
callback=record_result
)
print(" 4. 高优先级：high_1")
# 查看队列大小
print(f"\n队列大小：{scheduler.get_queue_size()}")
# 查看状态
status = scheduler.get_status()
print(f"初始令牌数：{status['token_bucket']['tokens']}")
print("\n✅ 优先级队列测试完成（仅提交，未处理）\n")
def test_retry_decorator():
"""测试重试装饰器"""
print("=" * 60)
print("测试 4: 重试装饰器")
print("=" * 60)
attempt_count = [0]
@retry_with_backoff(max_retries=3, base_delay=0.1, jitter=False)
def flaky_function():
attempt_count[0] += 1
if attempt_count[0] < 3:
raise Exception(f"模拟失败 (尝试 {attempt_count[0]})")
return f"成功 (尝试 {attempt_count[0]})"
print("\n调用易失败函数（前 2 次失败，第 3 次成功）...")
start = time.time()
result = flaky_function()
elapsed = time.time() - start
print(f"结果：{result}")
print(f"总尝试次数：{attempt_count[0]}")
print(f"总耗时：{elapsed:.3f}s")
print("\n✅ 重试装饰器测试完成\n")
def test_coordinated_poller():
"""测试统一轮询器"""
print("=" * 60)
print("测试 5: COO 统一轮询器（简化版，短间隔测试）")
print("=" * 60)
scheduler = RequestScheduler(rate=40/60, capacity=40)
poller = CoordinatedPoller(scheduler, poll_interval=2) # 2 秒轮询一次（测试用）
received_results = []
def on_poll_result(result):
received_results.append((datetime.now().strftime("%H:%M:%S"), result))
print(f" [{datetime.now().strftime('%H:%M:%S')}] 收到轮询结果")
poller.subscribe(on_poll_result)
print("\n启动轮询器（轮询间隔 2 秒，运行 5 秒后停止）...")
poller.start()
# 等待 5 秒
time.sleep(5)
poller.stop()
print(f"\n收到结果次数：{len(received_results)}")
for ts, result in received_results:
print(f" {ts}: {result['timestamp'][:19]}")
print("\n✅ 统一轮询器测试完成\n")
def test_rate_limit_stress():
"""压力测试：快速提交大量请求"""
print("=" * 60)
print("测试 6: 压力测试（40 RPM 限制下提交 50 个请求）")
print("=" * 60)
scheduler = RequestScheduler(rate=40/60, capacity=40, enable_cache=True)
scheduler.start()
completed = []
failed = []
lock = threading.Lock()
def callback(data):
with lock:
completed.append(data)
return data
print("\n快速提交 50 个请求...")
start_time = time.time()
for i in range(50):
priority = Priority.NORMAL if i % 10 != 0 else Priority.URGENT
scheduler.submit(
payload={"index": i},
priority=priority,
callback=callback
)
print("提交完成，等待处理...")
# 等待 10 秒
time.sleep(10)
elapsed = time.time() - start_time
# 查看统计
status = scheduler.get_status()
print(f"\n耗时：{elapsed:.2f}s")
print(f"队列大小：{status['queue_size']}")
print(f"已完成：{status['stats']['completed_requests']}")
print(f"失败：{status['stats']['failed_requests']}")
print(f"降级：{status['stats']['fallback_requests']}")
print(f"令牌桶状态：{status['token_bucket']}")
scheduler.stop()
print("\n✅ 压力测试完成\n")
def main():
"""运行所有测试"""
print("\n")
print("╔" + "=" * 58 + "╗")
print("║" + " " * 58 + "║")
print("║" + " BIZ-26 限流器测试套件".center(58) + "║")
print("║" + " API 请求优先级队列 + 令牌桶限流".center(58) + "║")
print("║" + " " * 58 + "║")
print("╚" + "=" * 58 + "╝")
print()
try:
test_token_bucket()
test_cache_manager()
test_priority_queue()
test_retry_decorator()
test_coordinated_poller()
test_rate_limit_stress()
print("\n")
print("╔" + "=" * 58 + "╗")
print("║" + " " * 58 + "║")
print("║" + " ✅ 所有测试完成".center(58) + "║")
print("║" + " " * 58 + "║")
print("╚" + "=" * 58 + "╝")
print()
except KeyboardInterrupt:
print("\n\n⚠️ 测试被用户中断\n")
except Exception as e:
print(f"\n\n❌ 测试出错：{e}\n")
import traceback
traceback.print_exc()
sys.exit(1)
if __name__ == "__main__":
main()