使用并发组限制每个方法的并发度

除了设置 Actor 的整体最大并发度外，Ray 还允许将方法分离到并发组中，每个组拥有自己的线程。这使你能够限制每个方法的并发度，例如，允许健康检查方法拥有独立于请求服务方法的并发配额。

提示

并发组适用于 asyncio 和线程化的 Actor。语法是相同的。

定义并发组

这定义了两个并发组：“io” 最大并发度为 2，“compute” 最大并发度为 4。方法 f1 和 f2 被放入“io”组，方法 f3 和 f4 被放入“compute”组。请注意，Actor 始终有一个默认并发组，对于 AsyncIO Actor，其默认并发度为 1000，否则为 1。

Python

你可以使用 concurrency_group 装饰器参数为 Actor 定义并发组

import ray

@ray.remote(concurrency_groups={"io": 2, "compute": 4})
class AsyncIOActor:
    def __init__(self):
        pass

    @ray.method(concurrency_group="io")
    async def f1(self):
        pass

    @ray.method(concurrency_group="io")
    async def f2(self):
        pass

    @ray.method(concurrency_group="compute")
    async def f3(self):
        pass

    @ray.method(concurrency_group="compute")
    async def f4(self):
        pass

    async def f5(self):
        pass

a = AsyncIOActor.remote()
a.f1.remote()  # executed in the "io" group.
a.f2.remote()  # executed in the "io" group.
a.f3.remote()  # executed in the "compute" group.
a.f4.remote()  # executed in the "compute" group.
a.f5.remote()  # executed in the default group.

Java

你可以使用 API 参数 setConcurrencyGroups() 为并发 Actor 定义并发组

class ConcurrentActor {
    public long f1() {
        return Thread.currentThread().getId();
    }

    public long f2() {
        return Thread.currentThread().getId();
    }

    public long f3(int a, int b) {
        return Thread.currentThread().getId();
    }

    public long f4() {
        return Thread.currentThread().getId();
    }

    public long f5() {
        return Thread.currentThread().getId();
    }
}

ConcurrencyGroup group1 =
    new ConcurrencyGroupBuilder<ConcurrentActor>()
        .setName("io")
        .setMaxConcurrency(1)
        .addMethod(ConcurrentActor::f1)
        .addMethod(ConcurrentActor::f2)
        .build();
ConcurrencyGroup group2 =
    new ConcurrencyGroupBuilder<ConcurrentActor>()
        .setName("compute")
        .setMaxConcurrency(1)
        .addMethod(ConcurrentActor::f3)
        .addMethod(ConcurrentActor::f4)
        .build();

ActorHandle<ConcurrentActor> myActor = Ray.actor(ConcurrentActor::new)
    .setConcurrencyGroups(group1, group2)
    .remote();

myActor.task(ConcurrentActor::f1).remote();  // executed in the "io" group.
myActor.task(ConcurrentActor::f2).remote();  // executed in the "io" group.
myActor.task(ConcurrentActor::f3, 3, 5).remote();  // executed in the "compute" group.
myActor.task(ConcurrentActor::f4).remote();  // executed in the "compute" group.
myActor.task(ConcurrentActor::f5).remote();  // executed in the "default" group.

默认并发组

默认情况下，方法被放置在默认并发组中，该组对于 AsyncIO Actor 的并发限制为 1000，否则为 1。可以通过设置 max_concurrency Actor 选项来更改默认组的并发度。

Python

以下 Actor 有 2 个并发组：“io” 和“default”。“io”的最大并发度为 2，“default”的最大并发度为 10。

@ray.remote(concurrency_groups={"io": 2})
class AsyncIOActor:
    async def f1(self):
        pass

actor = AsyncIOActor.options(max_concurrency=10).remote()

Java

以下并发 Actor 有 2 个并发组：“io” 和“default”。“io”的最大并发度为 2，“default”的最大并发度为 10。

class ConcurrentActor:
    public long f1() {
        return Thread.currentThread().getId();
    }

ConcurrencyGroup group =
    new ConcurrencyGroupBuilder<ConcurrentActor>()
        .setName("io")
        .setMaxConcurrency(2)
        .addMethod(ConcurrentActor::f1)
        .build();

ActorHandle<ConcurrentActor> myActor = Ray.actor(ConcurrentActor::new)
      .setConcurrencyGroups(group1)
      .setMaxConcurrency(10)
      .remote();

在运行时设置并发组

你还可以在运行时将 Actor 方法分派到特定的并发组。

以下代码片段演示了在运行时动态设置方法 f2 的并发组。

Python

你可以使用 .options 方法。

# Executed in the "io" group (as defined in the actor class).
a.f2.options().remote()

# Executed in the "compute" group.
a.f2.options(concurrency_group="compute").remote()

Java

你可以使用 setConcurrencyGroup 方法。

// Executed in the "io" group (as defined in the actor creation).
myActor.task(ConcurrentActor::f2).remote();

// Executed in the "compute" group.
myActor.task(ConcurrentActor::f2).setConcurrencyGroup("compute").remote();