feat: openapi as work design doc

Author: muverystrong

docs/design/openapi-as-worker.md

@@ -0,0 +1,240 @@
+# openapi作为bella-queue Worker - 设计文档
+
+## 1. 概述
+
+openapi作为 bella-queue Worker 的一个具体实现，通过充分利用渠道闲时余量来处理排队任务，基于实时容量指标动态调整任务消费，在不影响在线流量的前提下最大化渠道利用率。
+
+### 目标
+- 能够较准确评估渠道的实时负载,基于负载容量动态调整消费任务
+## 2. 设计
+
+### 主流程
+
+WorkerManager 作为核心协调组件，负责管理 Worker 生命周期和任务执行流程：
+
+#### 2.1 流程概述
+
+1. **Worker 初始化**
+   - 定期（每10分钟）扫描可用渠道，刷新 Worker 列表（剔除不可用、非queue_mode为pull的渠道，新增符合条件的渠道）
+   - 为每个渠道创建对应的 WorkerContext 实例，封装 Worker、CapacityCalculator 和 TaskProcessor
+
+2. **Worker 启动**
+   - WorkerContext 启动 BackoffTask
+   - BackoffTask 运行在单独的线程中，负责任务轮询和退避策略
+
+3. **任务执行轮询循环**
+   - **容量检查**：CapacityCalculator 基于历史429错误记录评估渠道最大容量
+   - **条件执行**：仅在余量大于一定比例时（默认70%）时调用 `worker.takeAndRun(take)`
+   - **连续处理**：通过 do-while 循环持续获取任务（每次1个）直到队列为空
+   - **委托执行**：TaskProcessor 处理具体任务，转换协议格式并委托给 Channel 处理
+   - **退避策略**：BackoffTask 实现指数退避，成功时减少间隔，失败时增加间隔
+
+#### 2.2 决策流程
+
+```
+┌─────────────────┐
+│   BackoffTask   │
+│   轮询循环      │
+└─────────┬───────┘
+          │
+          ▼
+┌─────────────────┐      ┌─────────────────┐
+│ CapacityCalculator│─────▶│   余量 > 70%? │
+│  评估渠道容量    │      └─────────┬───────┘
+└─────────────────┘                │
+                                   │
+                    ┌──────────────┼──────────────┐
+                    │ 是           │ 否           │
+                    ▼              ▼              
+          ┌─────────────────┐  ┌─────────────────┐
+          │   批量获取任务   │  │   跳过本轮      │
+          │  takeAndRun()   │  │  退避等待       │
+          └─────────┬───────┘  └─────────────────┘
+                    │
+                    ▼
+          ┌─────────────────┐
+          │  TaskProcessor  │
+          │  协议转换+执行   │
+          └─────────┬───────┘
+                    │
+                    ▼
+          ┌─────────────────┐
+          │ 处理响应状态码   │
+          │ 200:完成        │
+          │ 429/503:重试    │
+          │ 其他:失败       │
+          └─────────┬───────┘
+                    │
+                    ▼
+          ┌─────────────────┐
+          │ 记录Metrics     │
+          │ (429错误更新容量)│
+          └─────────────────┘
+```
+
+### 组件交互
+
+```
+┌─────────────────┐
+│  WorkerManager  │ ──── 每10分钟刷新Workers
+│                 │ ──── 管理WorkerContext生命周期
+└─────────┬───────┘        
+          │ 创建和管理               
+          ▼                
+┌──────────────────────────────┐
+│     WorkerContext            │ ──── 封装单个渠道的Worker组件
+│  - Worker (任务消费)          │ ──── 启动BackoffTask
+│  - CapacityCalculator(容量)   │ ──── 协调各组件交互
+│  - TaskProcessor (处理器)     │
+└──────────┬───────────────────┘
+           │
+           ▼
+┌──────────────────────────────┐
+│       BackoffTask            │ ──── 退避策略轮询
+│                              │ ──── 容量检查
+│                              │ ──── 任务批量获取  
+└──────────┬───────────────────┘
+           │
+           ▼
+┌──────────────────────────────┐
+│   CapacityCalculator         │ ──── 基于429历史数据
+│                              │ ──── EMA算法拟合容量  
+│                              │ ──── 本地缓存(5分钟)
+└──────────┬───────────────────┘
+           │
+           │ Redis存储
+           ▼
+┌──────────────────────────────┐
+│   Redis (429历史记录)        │ ──── rpm_429_history指标
+│                              │ ──── 时间戳+RPM+响应时间
+│                              │ ──── 支持多种拟合算法
+└──────────────────────────────┘
+```
+
+
+
+### 基于429历史的渠道容量评估
+
+通过收集渠道历史429限流错误时的RPM数据，使用拟合算法预测渠道的最大处理能力，为Worker任务消费决策提供依据。
+
+#### 1. 容量计算核心组件
+
+**CapacityCalculator** 负责渠道容量计算，主要包含：
+- **本地缓存**：5分钟缓存避免频繁计算
+- **历史数据获取**：从Redis读取`rpm_429_history`指标
+- **拟合算法**：支持可插拔的容量计算算法
+
+#### 2. 429历史数据结构
+
+系统在渠道出现429限流错误时记录关键指标到Redis:
+
+**存储Key**: `bella-openapi-channel-metrics:{channelCode}:rpm_429_history`
+
+**数据格式**:
+```json
+{
+  "timestamp": 1634567890000,
+  "rpm": 485.0,
+  "avg_response_time": 60000.0
+}
+```
+
+#### 3. EMA拟合算法实现
+
+**EmaFittingAlgorithm** 作为默认容量计算算法：
+
+**算法特性**:
+- **指数移动平均**：默认α=0.3，平滑处理历史RPM数据
+- **响应时间惩罚**：响应时间越长，容量估算越保守  
+- **向下取整**：确保容量估算不会过于乐观
+
+#### 4. 容量评估示例
+
+**场景1: 充足历史数据**
+```
+历史429记录:
+  [时间戳1: RPM=400, 响应时间=5s]
+  [时间戳2: RPM=450, 响应时间=8s] 
+  [时间戳3: RPM=380, 响应时间=12s]
+
+EMA计算:
+  RPM_EMA = 0.3*380 + 0.7*(0.3*450 + 0.7*400) = 408
+  ResponseTime_EMA = 0.3*12000 + 0.7*(0.3*8000 + 0.7*5000) = 8600ms
+  
+响应时间惩罚 = min(1.0, 10000/8600) = 1.0
+最终容量 = floor(408 * 1.0 + 0.5) = 408
+```
+
+**场景2: 响应时间过长惩罚**
+```
+历史429记录:
+  [时间戳1: RPM=500, 响应时间=25s]
+  
+EMA计算:
+  RPM_EMA = 500
+  ResponseTime_EMA = 25000ms
+  
+响应时间惩罚 = min(1.0, 10000/25000) = 0.4
+最终容量 = floor(500 * 0.4 + 0.5) = 200
+```
+
+**场景3: 无历史数据**
+```
+历史429记录: 空
+返回容量: 取得当前rpm值
+```
+
+## 3. 退避策略详细设计
+
+### 3.1 退避策略概述
+
+BackoffTask 实现智能退避机制，根据任务获取成功/失败情况动态调整轮询间隔，避免无效轮询的同时确保及时处理新任务。
+
+### 3.2 退避参数配置
+#### 退避策略
+- **有任务时**: 保持最小间隔(5ms)持续轮询，确保及时处理
+- **无任务时**: 从5秒开始退避，按1.5倍增长到最大5分钟
+- **最大等待后**: 达到5分钟后保持该间隔，直到再次获取到任务
+- **任务恢复**: 一旦有任务，立即重置到最小间隔持续轮询
+
+### 3.3 双重退避策略
+
+BackoffTask 实现两种不同的退避策略，根据失败原因选择合适的等待机制：
+
+#### 容量不足退避（基于响应时间）
+- **触发条件**: `余量 <= 70%`
+- **等待时间**: 使用渠道平均响应时间
+
+#### 任务队列空退避
+- **触发条件**: 有余量但 `takeAndRun()` 返回 0
+- **调整策略**: 
+  - 有任务时：立即重置到最小间隔(5ms)持续轮询
+  - 无任务时：从5秒开始按1.5倍增长到5分钟
+- **最大等待**: 达到5分钟后保持该间隔不再增长
+- **立即响应**: 一旦检测到任务，立即返回高频轮询模式
+
+## 4. 计费逻辑集成
+
+### 4.2 现有计费系统架构
+#### 计费流程
+```
+┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
+│  TaskProcessor  │───▶│  EndpointLogger │───▶│ Disruptor队列   │
+│  成功处理任务   │    │  logger.log()   │    │ 异步事件处理    │
+└─────────────────┘    └─────────────────┘    └─────────┬───────┘
+                                                        │
+┌─────────────────┐    ┌─────────────────┐    ┌─────────▼───────┐
+│   数据库更新    │◀───│   CostCounter   │◀───│  CostLogHandler │
+│ 原子性费用累计  │    │  内存累计(1分钟) │    │  计算具体费用   │
+└─────────────────┘    └─────────────────┘    └─────────────────┘
+```
+
+### 4.3 Worker计费集成实现
+在TaskProcessor的`executeTask`调用具体能力点的处理方法，拿到响应后,同步计费日志：
+```java
+EndpointProcessData processData = EndpointContext.getProcessData();
+processData.setResponse(response);
+logger.log(processData);
+```
+## 5. 如何防止在线(blocking, streaming)饿死
+问题出现在有一段时间，无法知道系统的实际容量，或者容量估算不准确。