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

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

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

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

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

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

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

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

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

task_priority:reply_customer:10# 最高campaign_signup:9upload_product:5collect_data:3# 最低``` 但很快就发现问题：同样是 `reply_customer`，回复一个刚刚投诉的客户和回复一个三天前的咨询，紧急程度完全不同。 纯静态优先级无法体现任务内部的差异。 我们改进为**基础优先级+ 动态加权**的模式。最终优先级分数由多个因子加权计算： ```pythondef 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 asyncioclass 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.5urgency = task.get("urgency",0) retry_bonus = task.get("retry_count",0) * 0.3 return base + wait_bonus + urgency + retry_bonusasync 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. 高优任务完成后，被挂起的低优任务恢复执行 ```pythonclass 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分钟，无论其优先级多低，都会被强制提升至可执行队列的前端，并标记为“不可被抢占”。 ```pythonasync def aging_check_loop(self):while True:# 查找等待超过30分钟的任务cutoff = time.time()-1800aged_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` 插件： ```pythonclass 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次抢占。超出则拒绝新的抢占请求，改为排队等待。---## 九、写在最后自动化系统的任务调度，不是简单的“先来后到”。 当业务有轻重缓急之分时，技术架构必须能够识别、响应、保障。 动态优先级和资源抢占，正是把“业务意识”注入到调度系统的一种方式。>自动化的目标不是做所有的事，而是在正确的时间，做最正确的事。---*作者：林焱*