import ray
import asyncio
@ray.remote
class AsyncActor:
    # multiple invocation of this method can be running in
    # the event loop at the same time
    async def run_concurrent(self):
        print("started")
        await asyncio.sleep(2) # concurrent workload here
        print("finished")
actor = AsyncActor.remote()
# regular ray.get
ray.get([actor.run_concurrent.remote() for _ in range(4)])
# async ray.get
async def async_get():
    await actor.run_concurrent.remote()
asyncio.run(async_get())
(AsyncActor pid=40293) started
(AsyncActor pid=40293) started
(AsyncActor pid=40293) started
(AsyncActor pid=40293) started
(AsyncActor pid=40293) finished
(AsyncActor pid=40293) finished
(AsyncActor pid=40293) finished
(AsyncActor pid=40293) finished
ObjectRefs 作为 asyncio.Futures#
ObjectRefs 可以转换为 asyncio.Futures。这一特性使得在现有的并发应用程序中可以 await ray futures。
import ray
@ray.remote
def some_task():
    return 1
ray.get(some_task.remote())
ray.wait([some_task.remote()])
你可以这样做：
import ray
import asyncio
@ray.remote
def some_task():
    return 1
async def await_obj_ref():
    await some_task.remote()
    await asyncio.wait([some_task.remote()])
asyncio.run(await_obj_ref())
更多关于 asyncio 的模式，包括超时和 asyncio.gather，请参阅 asyncio 文档。
如果你需要直接访问未来对象，你可以调用：
import asyncio
async def convert_to_asyncio_future():
    ref = some_task.remote()
    fut: asyncio.Future = asyncio.wrap_future(ref.future())
    print(await fut)
asyncio.run(convert_to_asyncio_future())
ObjectRefs 作为 concurrent.futures.Futures#
ObjectRefs 也可以被包装成 concurrent.futures.Future 对象。这对于与现有的 concurrent.futures API 接口非常有用：
import concurrent
refs = [some_task.remote() for _ in range(4)]
futs = [ref.future() for ref in refs]
for fut in concurrent.futures.as_completed(futs):
    assert fut.done()
    print(fut.result())
    async def run_task(self):
        print("started")
        await asyncio.sleep(2) # Network, I/O task here
        print("ended")
actor = AsyncActor.remote()
# All 5 tasks should start at once. After 2 second they should all finish.
# they should finish at the same time
ray.get([actor.run_task.remote() for _ in range(5)])
(AsyncActor pid=3456) started
(AsyncActor pid=3456) started
(AsyncActor pid=3456) started
(AsyncActor pid=3456) started
(AsyncActor pid=3456) started
(AsyncActor pid=3456) ended
(AsyncActor pid=3456) ended
(AsyncActor pid=3456) ended
(AsyncActor pid=3456) ended
(AsyncActor pid=3456) ended
在底层，Ray 在单个 Python 事件循环中运行所有方法。请注意，不允许在异步 actor 方法中运行阻塞的 ray.get 或 ray.wait，因为 ray.get 会阻塞事件循环的执行。
在异步角色中，任何时候只能运行一个任务（尽管任务可以多路复用）。AsyncActor 中将只有一个线程！如果你想要一个线程池，请参阅 线程化角色。
在异步执行者中设置并发#
你可以使用 max_concurrency 标志设置一次运行的“并发”任务数量。默认情况下，可以同时运行1000个任务。
import asyncio
@ray.remote
class AsyncActor:
    async def run_task(self):
        print("started")
        await asyncio.sleep(1) # Network, I/O task here
        print("ended")
actor = AsyncActor.options(max_concurrency=2).remote()
# Only 2 tasks will be running concurrently. Once 2 finish, the next 2 should run.
ray.get([actor.run_task.remote() for _ in range(8)])
(AsyncActor pid=5859) started
(AsyncActor pid=5859) started
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) started
(AsyncActor pid=5859) started
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) started
(AsyncActor pid=5859) started
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) started
(AsyncActor pid=5859) started
(AsyncActor pid=5859) ended
(AsyncActor pid=5859) ended
线程化角色#
有时，asyncio 并不是你角色的理想解决方案。例如，你可能有一个方法执行一些计算密集型任务，同时阻塞事件循环，不通过 await 放弃控制。这会损害异步角色的性能，因为异步角色一次只能执行一个任务，并且依赖 await 进行上下文切换。
相反，你可以使用 max_concurrency 演员选项，而无需任何异步方法，从而实现线程并发（如线程池）。
当actor定义中至少有一个 async def 方法时，Ray 会将该actor识别为 AsyncActor 而不是 ThreadedActor。
@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()])
(ThreadedActor pid=4822) I'm running in a thread!
(ThreadedActor pid=4822) I'm running in another thread!
每个线程化角色的调用都将在一个线程池中运行。线程池的大小受 max_concurrency 值的限制。
远程任务的AsyncIO#
我们不支持远程任务的 asyncio。以下代码片段将会失败：
@ray.remote
async def f():
相反，你可以用一个包装器包裹 async 函数来同步运行任务：
async def f():
@ray.remote
def wrapper():
    import asyncio
    asyncio.run(f())