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

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

注意

您还可以使用 threaded actors 来实现并发。

用例

您有一个 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(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 方法会永远运行，并且不会将控制权让给其他方法。我们可以通过使用 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 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}")

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