模式：使用 asyncio 并发运行 actor 方法

默认情况下，Ray actor 在单个线程中运行，actor 方法调用按顺序执行。这意味着长时间运行的方法调用会阻塞所有后续调用。在此模式中，我们使用 await 从长时间运行的方法调用中让出控制权，以便其他方法调用可以并发运行。通常在方法执行 IO 操作时让出控制权，但你也可以使用 await asyncio.sleep(0) 显式让出控制权。

注意

你也可以使用 线程化 actor 来实现并发。

示例用例

你有一个 actor，其中包含一个长时间轮询的方法，该方法不断从远程存储中获取任务并执行。你还希望在长时间轮询方法运行时查询已执行任务的数量。

使用默认 actor，代码看起来像这样

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(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 方法会永远运行，并且从不让出控制权来运行其他方法。我们可以通过使用 async actors 并使用 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 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}")

在这里，我们没有使用阻塞的 ray.get() 来获取 ObjectRef 的值，而是使用 await，这样在我们等待对象被获取时，它可以让出控制权。