企业官网建设流程全解析

Ray 并发模式实战：用 Async Actor（asyncio）让 Actor 方法并发执行

【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray

默认情况下，Ray 的 Actor 在单个线程中串行执行方法调用，一个长时间运行的方法会阻塞其后的所有调用。本指南基于 Ray 官方 Patterns 文档，讲解如何利用 asyncio 将 Actor 改造为 Async Actor，通过await主动让出控制权，让长轮询、I/O 密集型方法与其他查询类方法在同一进程内并发执行。读完本文你将掌握同步 Actor 的阻塞问题诊断、Async Actor 的改造方法、max_concurrency并发控制，以及它与 Threaded Actor 的选型取舍。

问题背景：单线程 Actor 的顺序执行

Ray 的普通 Actor（详见 Actor 基础文档）默认运行在单个线程中，其方法调用严格按照提交顺序串行执行。这意味着：

• 一个执行时间很长的方法会阻塞所有后续提交的方法；
• 即使后续方法只是简单的状态查询，也必须等待长任务结束后才能运行；
• 单个 Actor 内部无法通过多方法调用获得任何并发能力。

这种模型的好处是状态访问天然线程安全，但代价是当 Actor 内部存在"永不返回"的长运行方法时，整个 Actor 将失去响应能力。

示例场景：长轮询取任务 + 实时查询进度

原文档给出的典型场景是：

• 一个 Actor 内部有长轮询方法，持续不断地从远端存储获取任务并执行；
• 与此同时，用户希望随时查询该 Actor 已经执行的任务数量。

在默认的同步 Actor 下，长轮询方法一旦启动就占据整个线程，查询方法永远得不到执行机会。仓库中的完整示例代码位于 pattern_async_actor.py。

同步版本：ray.get阻塞导致的方法饿死

先看同步实现。TaskStore负责产出任务，TaskExecutor负责拉取并执行，同时维护一个执行计数：

import ray @ray.remote class TaskStore: def get_next_task(self): return "task" @ray.remote class TaskExecutor: def __init__(self, task_store): self.task_store = task_store self.num_executed_tasks = 0 def run(self): while True: task = ray.get(self.task_store.get_next_task.remote()) self._execute_task(task) def _execute_task(self, task): # Executing the task self.num_executed_tasks = self.num_executed_tasks + 1 def get_num_executed_tasks(self): return self.num_executed_tasks task_store = TaskStore.remote() task_executor = TaskExecutor.remote(task_store) task_executor.run.remote() try: # This will timeout since task_executor.run occupies the entire actor thread # and get_num_executed_tasks cannot run. ray.get(task_executor.get_num_executed_tasks.remote(), timeout=5) except ray.exceptions.GetTimeoutError: print("get_num_executed_tasks didn't finish in 5 seconds")

这里的问题非常明确：TaskExecutor.run中的while True循环永远运行，且ray.get(self.task_store.get_next_task.remote())是阻塞调用，整个 Actor 线程被它独占。get_num_executed_tasks提交后永远无法获得执行机会，ray.get(..., timeout=5)最终抛出ray.exceptions.GetTimeoutError——这正好验证了"默认 Actor 方法严格串行"的行为。这段带超时的代码本身也是一个实用诊断技巧：用timeout参数探测方法是否被阻塞。

异步版本：用await让出控制权

解决思路是把 Actor 改造成Async Actor：将方法定义为async def，并把阻塞的ray.get替换为await一个 ObjectRef。await在等待远端结果期间会让出控制权给事件循环，使其他方法得以插队执行：

@ray.remote class AsyncTaskExecutor: def __init__(self, task_store): self.task_store = task_store self.num_executed_tasks = 0 async def run(self): while True: # Here we use await instead of ray.get() to # wait for the next task and it will yield # the control while waiting. task = await self.task_store.get_next_task.remote() self._execute_task(task) def _execute_task(self, task): # Executing the task self.num_executed_tasks = self.num_executed_tasks + 1 def get_num_executed_tasks(self): return self.num_executed_tasks async_task_executor = AsyncTaskExecutor.remote(task_store) async_task_executor.run.remote() # We are able to run get_num_executed_tasks while run method is running. num_executed_tasks = ray.get(async_task_executor.get_num_executed_tasks.remote()) print(f"num of executed tasks so far: {num_executed_tasks}")

改造点只有两处，但效果是质的改变：

1. def run(self)→async def run(self)：让 Ray 把该 Actor 识别为 Async Actor；
2. task = ray.get(...)→task = await self.task_store.get_next_task.remote()：在等待 ObjectRef 期间让出事件循环，而不是阻塞线程。

现在AsyncTaskExecutor.run虽然在无限循环中持续运行，但每次等待任务到达时都会通过await释放控制权，因此get_num_executed_tasks可以随时并发执行并返回当前进度。

主动让出控制权：await asyncio.sleep(0)

await通常发生在方法执行I/O 操作（如网络请求、读取远端存储）的时候，这是让出控制权最常见也最自然的时机。但如果你希望在没有真实 I/O 等待的代码段中显式让出控制权，可以使用await asyncio.sleep(0)。asyncio.sleep(0)会立即返回，但它会触发一次事件循环调度，把执行机会交给其他排队的协程，是 asyncio 中标准的"让出 CPU"惯用法。

原理：Ray 如何识别 Async Actor

Ray 并不是通过显式声明来区分 Async Actor 的，而是自动检测类中是否存在异步方法。相关实现在 python/ray/_private/async_compat.py：

def is_async_func(func) -> bool: """Return True if the function is an async or async generator method.""" return inspect.iscoroutinefunction(func) or inspect.isasyncgenfunction(func) @lru_cache(maxsize=2**10) def has_async_methods(cls: object) -> bool: """Return True if the class has any async methods.""" return len(inspect.getmembers(cls, predicate=is_async_func)) > 0

在创建 Actor 时，python/ray/actor.py 会调用has_async_methods判定类型，并据此设置默认并发参数：

is_asyncio = has_async_methods(meta.modified_class) if actor_options.get("max_concurrency") is None: actor_options["max_concurrency"] = ( ... 1000 # for asyncio execution ... )

也就是说，只要类中存在至少一个async def方法，Ray 就会把该 Actor 视为 Async Actor，其内部方法将运行在同一个 asyncio 事件循环上。底层 C++ 端（src/ray/core_worker/context.cc）也会在 actor 任务执行时记录current_actor_is_asyncio_与current_actor_max_concurrency_，用于调度层面的并发控制。

Async Actor 的关键语义（详见 AsyncIO / Concurrency for Actors）：

• 所有方法运行在单个 Python 事件循环中，只有一个线程；
• 同一时刻只有一个任务在真正执行，任务之间通过await进行多路复用（multiplexed）；
• 在 async 方法内禁止使用阻塞的ray.get或ray.wait，因为它们会卡住整个事件循环，导致所有方法失去响应。

并发上限：max_concurrency选项

Async Actor 默认允许最多1000个方法调用"同时"排队运行（实际执行仍是事件循环交替进行的）。你可以通过.options(max_concurrency=...)限制并发数，这常被用来控制资源占用或实现批处理语义。以官方文档中的批处理示例为参考：

actor = AsyncActor.options(max_concurrency=2).remote(2) # Only 2 tasks will run concurrently. # Once 2 finish, the next 2 should run. ray.get([actor.run_task.remote() for _ in range(8)])

max_concurrency=2时，8 个任务会以每批 2 个的方式进入并发执行。

关于max_concurrency，python/ray/actor.py 的 API 文档明确说明了几条重要约束：

• 它只对direct actor call（直连调用）生效；
• 默认值：threaded 执行为1，asyncio 执行为1000；
• 当max_concurrency > 1时，执行顺序不再保证；
• 使用多线程（max_concurrency > 1）或 Async Actor 时，allow_out_of_order_execution必须为True（默认即如此），因为并发执行天然会打乱提交顺序。

进阶：ObjectRef 与 asyncio.Future 的互操作

Async Actor 场景下，你还可以把 ObjectRef 直接当作 asyncio 可等待对象使用，这在已有异步代码中集成 Ray 时非常方便（参考 async_api.rst 中的完整示例）：

import asyncio import ray @ray.remote def some_task(): return 1 async def await_obj_ref(): await some_task.remote() await asyncio.wait([some_task.remote()])

在 Python 3.11+ 上，还可以把 ObjectRef 包装成标准的asyncio.Future对象：

async def convert_to_asyncio_future(): ref = some_task.remote() fut: asyncio.Future = asyncio.wrap_future(ref.future()) print(await fut)

与 Threaded Actor 的选型对比

原文档特别提示：你同样可以使用 Threaded Actor 实现并发（详见 Threaded Actors）。两者适用场景不同：

关键判断依据：如果方法内存在无法通过await让出控制权的计算密集段，Async Actor 反而会被拖慢——因为事件循环只有这一个任务在跑，其他任务全部饿死。此时应改用 Threaded Actor，让长计算运行在独立线程中。

@ray.remote class ThreadedActor: def task_1(self): print("I'm running in a thread!") def task_2(self): print("I'm running in another thread!") a = ThreadedActor.options(max_concurrency=2).remote() ray.get([a.task_1.remote(), a.task_2.remote()])

注意一个容易踩坑的规则：只要 Actor 中存在一个async def方法，Ray 就会把它识别为 Async Actor 而非 Threaded Actor，因此不要混用两种模式。

另外需要清醒认识 Python 的 GIL 限制：无论 Async Actor 还是 Threaded Actor，同一时刻只有一个线程能执行 Python 字节码（详见 async_api.rst 的说明）。只有当代码调用 NumPy、Cython、TensorFlow、PyTorch 等会释放 GIL的原生库时，才能真正获得并行加速。这两种并发模型的价值在于避免阻塞、提升吞吐与响应性，而非突破 GIL。

补充：remote task 不支持 asyncio

需要特别区分的是，Ray 的remote task（无状态任务）不支持 asyncio：直接定义@ray.remote async def f()会失败。如果确实需要在任务中运行异步代码，可以包一层同步 wrapper：

async def f(): pass @ray.remote def wrapper(): import asyncio asyncio.run(f())

也就是说，asyncio 集成是 Actor 专属能力，与普通任务不同。

总结

本文核心模式可归纳为三步：

1. 识别阻塞点：默认同步 Actor 中，长运行方法（尤其是while True轮询 + 阻塞ray.get）会饿死后续方法；
2. 异步化改造：将方法改为async def，用await object_ref替代ray.get，在等待 I/O 时让出控制权；需要主动让出时使用await asyncio.sleep(0)；
3. 按需控制并发：用max_concurrency调节并发上限，并在计算密集场景改用 Threaded Actor。

这套模式非常适合"后台持续干活 + 前台随时查询"的 Actor 设计，例如任务队列消费者、监控探针、模型批处理 worker 等。完整的可运行示例与测试代码见 pattern_async_actor.py，更系统的 asyncio/并发说明可继续阅读 AsyncIO / Concurrency for Actors。

【免费下载链接】rayRay is an AI compute engine. Ray consists of a core distributed runtime and a set of AI Libraries for accelerating ML workloads.项目地址: https://gitcode.com/gh_mirrors/ra/ray