实用类

演员池

Python

ray.util 模块包含一个实用类，ActorPool。这个类类似于 multiprocessing.Pool，允许你在固定数量的执行者池上调度 Ray 任务。

import ray
from ray.util import ActorPool


@ray.remote
class Actor:
    def double(self, n):
        return n * 2


a1, a2 = Actor.remote(), Actor.remote()
pool = ActorPool([a1, a2])

# pool.map(..) returns a Python generator object ActorPool.map
gen = pool.map(lambda a, v: a.double.remote(v), [1, 2, 3, 4])
print(list(gen))
# [2, 4, 6, 8]

更多信息请参阅 包参考。

Java

Java 中尚未实现 Actor 池。

C++

C++ 中尚未实现 Actor 池。

使用 Ray 队列进行消息传递

有时仅使用一个信号进行同步是不够的。如果你需要在多个任务或角色之间发送数据，你可以使用 ray.util.queue.Queue。

import ray
from ray.util.queue import Queue, Empty

ray.init()
# You can pass this object around to different tasks/actors
queue = Queue(maxsize=100)


@ray.remote
def consumer(id, queue):
    try:
        while True:
            next_item = queue.get(block=True, timeout=1)
            print(f"consumer {id} got work {next_item}")
    except Empty:
        pass


[queue.put(i) for i in range(10)]
print("Put work 1 - 10 to queue...")

consumers = [consumer.remote(id, queue) for id in range(2)]
ray.get(consumers)

Ray 的 Queue API 与 Python 的 asyncio.Queue 和 queue.Queue 具有类似的 API。