使用并发组按方法限制并发

除了为 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

您可以使用 setConcurrencyGroups() API 参数为并发 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(group)
      .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();