第274篇任务调度与编排——多任务机器人的执行管理

郑伟强dev

94人浏览 · 2026-08-27 06:51:10

郑伟强dev · 2026-08-27 06:51:10 发布

行为树系列三篇聊完了，从基础节点到设计模式再到和状态机的对比，行为树的核心知识应该清楚了。

但行为树解决的是"单个任务的执行逻辑"。实际项目中，机器人往往同时有多个任务要处理——送餐、巡检、回充、接待客人。这些任务之间有优先级、有依赖关系、有资源竞争。怎么决定"现在做哪个任务"？这就是任务调度的问题。

任务调度是行为树的"上级"。行为树管"怎么做"，任务调度管"做什么"。两者配合才能构成完整的任务管理方案。

一、任务调度的核心问题

多任务机器人面临的核心问题就三个：

优先级——多个任务同时到来，先做哪个？回充和送餐同时需要执行，电量低于10%时回充优先，电量30%时送餐优先。

资源竞争——多个任务需要同一资源（比如底盘），怎么协调？导航和搬运都需要底盘，不能同时执行。

任务依赖——某些任务必须在其他任务完成后才能开始？比如"搬运到A点"完成后才能"在A点安装"。

这三个问题在单机器人场景下还算好处理，到了多机器人场景就更复杂了。

二、任务队列与优先级调度

最简单的方案是任务队列。所有任务按优先级排队，每次从队头取一个执行。

import heapq
from dataclasses import dataclass, field

@dataclass
class Task:
    priority: int      # 越小越优先
    name: str
    params: dict
    order: int = field(default=0)  # 同优先级按到达顺序
    
class TaskScheduler:
    def __init__(self):
        self.queue = []
        self.counter = 0
    
    def add_task(self, priority, name, params):
        task = Task(priority, name, params, self.counter)
        self.counter += 1
        heapq.heappush(self.queue, (priority, task.order, task))
    
    def get_next_task(self):
        if self.queue:
            _, _, task = heapq.heappop(self.queue)
            return task
        return None

这个方案简单但有效。大部分仓储AGV的调度系统就是这么做的——任务有固定的优先级（回充>紧急避障>搬运>空闲），按优先级排队执行。

问题在于：优先级不是固定的。电量低时回充优先级最高，电量充足时搬运优先级最高。这就需要动态优先级。

动态优先级的实现也不复杂——每次取任务前重新计算所有待执行任务的有效优先级，而不是用固定的优先级值。

def get_effective_priority(task, robot_state):
    base = task.base_priority
    if task.name == "charge" and robot_state.battery < 20:
        return 0  # 紧急回充，最高优先级
    if task.name == "delivery" and task.urgency == "high":
        return 1  # 紧急配送
    return base

三、动态优先级与抢占式调度

动态优先级的意思是：任务的优先级根据当前状态实时调整。

class DynamicScheduler:
    def __init__(self):
        self.tasks = []
        self.current_task = None
    
    def compute_priority(self, task):
        base = task.base_priority
        # 电量低时提升回充优先级
        if task.name == "charge" and self.battery < 20:
            return 0  # 最高优先级
        # 有客人等待时提升接待优先级
        if task.name == "reception" and self.has_waiting_guest:
            return 1
        return base
    
    def schedule(self):
        # 检查是否需要抢占当前任务
        for task in self.tasks:
            p = self.compute_priority(task)
            if self.current_task and p < self.current_task.priority:
                # 新任务优先级更高，抢占
                self.preempt_current(task)
                return task
        return self.get_next_from_queue()

抢占式调度的关键问题是：被抢占的任务怎么办？

可恢复的——暂停当前任务，等抢占任务完成后恢复。比如导航被回充抢占，充满电后继续导航。

可丢弃的——直接取消。比如"去某个位置巡检"被抢占后，这次巡检就不做了。

不可中断的——等当前动作完成后再切换。比如机械臂正在夹取物体，不能中途停下来。

四、任务编排：DAG执行

有些任务之间有依赖关系，不能简单排队。比如一个装配任务：

1. 搬运零件A到工位 → 2. 搬运零件B到工位 → 3. 组装
                                              ↓
                              4. 质检 → 5. 入库

零件A和零件B的搬运可以并行，但都必须在组装之前完成。组装完成后才能质检和入库。

这种有向无环图（DAG）结构的任务编排，用行为树不太方便表达。通常用专门的任务编排框架来处理，比如ROS2的launch系统或者自定义的DAG调度器。

class DAGScheduler:
    def __init__(self):
        self.graph = {}  # task -> [依赖它的任务]
        self.completed = set()
    
    def add_task(self, task_id, dependencies):
        self.graph[task_id] = dependencies
    
    def get_ready_tasks(self):
        ready = []
        for task_id, deps in self.graph.items():
            if task_id not in self.completed:
                if all(d in self.completed for d in deps):
                    ready.append(task_id)
        return ready
    
    def complete_task(self, task_id):
        self.completed.add(task_id)

DAG调度器每个tick检查哪些任务的前置依赖已经完成，把这些任务放入可执行队列。多个就绪任务可以并行执行（如果资源允许）。

实际项目中，DAG编排常和任务调度器结合使用。DAG定义任务之间的依赖关系，调度器负责在依赖满足的前提下按优先级执行。

举个仓储物流的例子：一个订单包含"拣货→打包→配送"三个步骤。拣货环节可能需要多台机器人协作（不同货架的货物同时拣），打包环节需要等拣货全部完成，配送环节需要等打包完成。整个流程就是一个DAG。

订单DAG: 拣货A(机器人1) ─┐ 拣货B(机器人2) ─┼→ 打包 → 配送拣货C(机器人3) ─┘

这种场景下，任务调度器需要同时管理DAG依赖和多机器人资源分配。