影刀RPA店群自动化教程：Python动态优先级队列与浏览器资源抢占实战

当“上货”和“回复客户”同时等待执行时，系统必须知道谁先谁后。

否则，一个无关紧要的数据采集任务，可能卡住一条差评回复整整十分钟。

店群运营中，任务的紧急程度天然不同。
客服回复慢一分钟，可能多一条差评；大促活动的报名晚十分钟，可能直接错过入口。
但常规的定时数据采集早五分钟晚五分钟，几乎没有影响。

然而在早期架构中，我们所有任务共用一套FIFO队列。
先提交的先执行，完全不管业务含义。
这导致了多次紧急任务被普通任务长时间阻塞的事故。

我们后来设计了一套动态优先级队列，并结合浏览器资源抢占机制，让高优任务能在几秒内拿到执行权。
这篇文章就完整复盘这套机制的工程实践。

一、任务优先级的定义：不是简单的高中低

最初我们给每个任务类型设了一个固定优先级，写死在配置文件里：

task_priority:
  reply_customer: 10    # 最高
    campaign_signup: 9
      upload_product: 5
        collect_data: 3       # 最低
        ```
但很快就发现问题：同样是 `reply_customer`，回复一个刚刚投诉的客户和回复一个三天前的咨询，紧急程度完全不同。  
纯静态优先级无法体现任务内部的差异。

我们改进为**基础优先级 + 动态加权**的模式。  
最终优先级分数由多个因子加权计算：

```python
def calculate_priority(task):
    base = PRIORITY_BASE.get(task.type, 5)
        # 等待时间加权：每等待1分钟加0.5分
            wait_minutes = (time.time() - task.created_at) / 60
                wait_bonus = wait_minutes * 0.5
                    # 业务紧急系数：如客服消息的剩余回复时间
                        urgency = task.urgency_factor or 0
                            # 失败重试加权：失败次数越多，优先级适当提升，避免饥饿
                                retry_bonus = task.retry_count * 0.3
                                    
                                        return base + wait_bonus + urgency + retry_bonus
                                        ```
随着等待时间增加，低优先级任务的分数会逐步上升，避免永远得不到执行。  
同时，高紧急系数（如差评回复窗口仅剩2分钟）能瞬间把任务推上队首。

---

## 二、基于Redis Sorted Set的优先级队列

优先级队列的实现选用了Redis的Sorted Set，成员为任务序列化字符串，分数即为优先级。

```python
import json
import time
import asyncio

class PriorityTaskQueue:
    def __init__(self, redis_client, queue_key="task:priority_queue"):
            self.redis = redis_client
                    self.queue_key = queue_key
    async def enqueue(self, task: dict):
            score = self._calculate_score(task)
                    # 使用 task_id + 微秒时间戳 保证成员唯一
                            member = json.dumps(task)
                                    await self.redis.zadd(self.queue_key, {member: score})
    def _calculate_score(self, task: dict) -> float:
            base = task.get("base_priority", 5)
                    wait_sec = time.time() - task.get("created_at", time.time())
                            wait_bonus = wait_sec / 120 * 0.5  # 每2分钟加0.5
                                    urgency = task.get("urgency", 0)
                                            retry_bonus = task.get("retry_count", 0) * 0.3
                                                    return base + wait_bonus + urgency + retry_bonus
    async def dequeue(self) -> dict:
            # 原子操作：取出最高优先级（分数最大）的任务
                    result = await self.redis.zpopmax(self.queue_key, count=1)
                            if result:
                                        return json.loads(result[0][0])
                                                return None
    async def peek_high_priority(self, threshold: float):
            # 查看分数高于阈值的任务，不取出
                    return await self.redis.zrangebyscore(
                                self.queue_key, min=threshold, max='+inf', withscores=True
                                        )
                                        ```
每次入队时重新计算优先级分数并更新Sorted Set。  
出队时使用 `ZPOPMAX`，保证取到的永远是当前分数最高的任务。

---

## 三、资源抢占：高优任务不能等

有了优先级队列，高优任务可以快速排到队首。  
但如果当前所有Worker的浏览器实例都被低优任务占满，高优任务仍然要等。

**真正的紧急任务，不应该排队。**

我们引入了浏览器资源抢占机制。  
高优任务（优先级分数超过某阈值）在入队时，会触发一个抢占检查：

1. 确定任务目标店铺的浏览器实例是否被占用
2. 2. 如果占用者是一个低优先级任务，且该任务处于可安全挂起的阶段（如页面加载中、执行非关键步骤），则强制挂起低优先级任务
3. 3. 低优先级任务状态保存到Redis，浏览器实例被释放并分配给高优任务
4. 4. 高优任务完成后，被挂起的低优任务恢复执行
```python
class ResourcePreemptor:
    PREEMPT_SCORE_THRESHOLD = 8.0  # 超过此分值的任务具备抢占资格
    def __init__(self, browser_pool, task_state_store):
            self.browser_pool = browser_pool
                    self.state_store = task_state_store
    async def try_preempt(self, shop_id: str, incoming_task: dict) -> bool:
            if incoming_task.get("priority_score", 0) < self.PREEMPT_SCORE_THRESHOLD:
                        return False
        current_holder = self.browser_pool.get_holder(shop_id)
                if current_holder is None:
                            return False  # 资源空闲，无需抢占
        holder_score = current_holder.get("priority_score", 0)
                if holder_score >= incoming_task["priority_score"]:
                            return False  # 当前任务优先级不低，不抢占
        if not current_holder.get("preemptible", False):
                    return False  # 当前任务处于不可中断的临界区
        # 执行抢占
                await self._suspend_task(current_holder)
                        self.browser_pool.force_release(shop_id)
                                logger.info(f"Preempted shop {shop_id} from task {current_holder['task_id']}")
                                        return True
    async def _suspend_task(self, task_info: dict):
            # 保存任务上下文：当前步骤、已执行指令索引、数据上下文
                    await self.state_store.save(task_info["task_id"], task_info["context"])
                            # 通知影刀流程暂停（通过gRPC信号）
                                    await self._send_pause_signal(task_info["task_id"])
                                    ```
低优先级任务被挂起时，会保存当前执行进度（如已完成的步骤序号、页面状态等）。  
被挂起的任务会在优先级老化或被抢占任务完成后重新恢复。

---

## 四、抢占的安全边界：不是所有任务都能打断

抢占有风险。如果低优任务正在执行写操作（如提交表单、支付），强制中断可能导致脏数据。

我们为影刀指令序列中的每个步骤定义了 `interruptible` 属性。

```json
{
  "steps": [
      {"action": "navigate", "interruptible": true},
          {"action": "type_text", "interruptible": true},
              {"action": "upload_file", "interruptible": true},
                  {"action": "click", "locator": "#submit-btn", "interruptible": false},
                      {"action": "wait_element", "locator": ".success-msg", "interruptible": true}
                        ]
                        }
                        ```
`interruptible: false` 的步骤是不可抢占的临界区。  
抢占检查时，如果当前执行步骤标记为不可中断，拒绝抢占。

这保证了“提交订单”、“确认支付”等关键操作不会被中途打断。

---

## 五、优先级老化与饥饿避免

单纯依赖优先级可能导致低优先级任务长期被“饿死”。  
虽然我们在计算分数时加入了等待时间加权，但还需要一个硬性兜底：

当某个任务在队列中等待超过30分钟，无论其优先级多低，都会被强制提升至可执行队列的前端，并标记为“不可被抢占”。

```python
async def aging_check_loop(self):
    while True:
            # 查找等待超过30分钟的任务
                    cutoff = time.time() - 1800
                            aged_tasks = await self.redis.zrangebyscore(
                                        self.queue_key, min='-inf', max='+inf', withscores=True
                                                )
                                                        for member, score in aged_tasks:
                                                                    task = json.loads(member)
                                                                                if task.get("created_at", 0) < cutoff:
                                                                                                # 提升优先级至最高，并标记不可抢占
                                                                                                                task["priority_boosted"] = True
                                                                                                                                task["preemptible"] = False
                                                                                                                                                new_score = 100.0
                                                                                                                                                                await self.redis.zadd(self.queue_key, {json.dumps(task): new_score})
                                                                                                                                                                        await asyncio.sleep(60)
                                                                                                                                                                        ```
这种机制保证了长时间等待的任务最终一定能获得执行机会，避免因业务优先级导致运维事故。

---

## 六、跨平台差异处理

不同平台的任务，紧急程度判定逻辑不同。

- 拼多多客服回复有48小时窗口，但差评回复窗口短
- - TEMU上货任务在审核期结束前1小时，紧急度骤升
- - TikTok Shop直播中商品讲解需要在特定时段完成
我们为每个平台实现了一个 `UrgencyEstimator` 插件：

```python
class UrgencyEstimator:
    def estimate(self, task: dict) -> float:
            platform = task.get("platform")
                    if platform == "pdd":
                                return self._pdd_urgency(task)
                                        elif platform == "temu":
                                                    return self._temu_urgency(task)
                                                            elif platform == "tiktok":
                                                                        return self._tiktok_urgency(task)
                                                                                return 0.0
    def _pdd_urgency(self, task):
            if task["type"] == "reply_customer":
                        remaining_sec = task.get("deadline") - time.time()
                                    if remaining_sec < 3600:  # 1小时内
                                                    return 5.0
                                                            return 0.0
                                                            ```
这样新的平台加入时，只需实现对应的紧急度评估逻辑，优先级系统保持通用。

---

## 七、监控与告警

优先级队列和抢占机制需要被监控，否则出了“饿死”或“过度抢占”的问题难以发现。

关键指标：
- 队列中最高优先级分数与最低优先级分数的差距
- - 过去1小时内抢占次数
- - 等待超过30分钟的任务数量
- - 被抢占任务的恢复成功率
Grafana看板上，这些指标与常规任务执行成功率并列展示。  
一旦抢占次数激增，说明系统中有大量紧急任务，可能意味着上游运营策略出了问题。

---

## 八、踩坑实录

**分数精度问题。**  
Redis Sorted Set使用双精度浮点数存储分数。当等待时间加权导致分数极接近时，可能无法区分优先级。  
我们将分数计算基值设为整数，加权部分放在小数部分，并用足够精度，保证了区分度。

**抢占后恢复失败。**  
被挂起的低优任务在恢复时，偶尔会因为浏览器页面状态发生变化而导致元素定位失败。  
我们在恢复时加入了一个“页面重置”步骤：回到任务初始页面，重新执行已完成的非写操作步骤，跳过已完成的写操作步骤。  
这依赖于之前实现的指令序列化和幂等设计。

**抢占风暴。**  
有一次运营批量创建了大量紧急任务，系统在短时间内发生了数十次抢占，导致整体吞吐量反而下降。  
我们加了硬性限制：每个Worker每分钟最多允许2次抢占。超出则拒绝新的抢占请求，改为排队等待。

---

## 九、写在最后

自动化系统的任务调度，不是简单的“先来后到”。

当业务有轻重缓急之分时，技术架构必须能够识别、响应、保障。  
动态优先级和资源抢占，正是把“业务意识”注入到调度系统的一种方式。

> 自动化的目标不是做所有的事，而是在正确的时间，做最正确的事。
---

*作者：林焱*