Azure Cosmos DB Java SDK v4 中的吞吐量控制组

吞吐量控制组允许限制Azure Cosmos DB操作的请求单位（RU）消耗量。 如果有多个应用程序或工作负荷共享同一容器，并且希望确保一个工作负荷不会消耗所有可用的吞吐量，这非常有用。 从版本 4.13.0 开始，Azure Cosmos DB Java SDK v4 中提供了吞吐量控制。

Java SDK 支持三种吞吐量控制模式：

• 本地吞吐量控制 - 限制单个客户端实例中的 RU 消耗。
• 全局吞吐量控制 - 通过元数据容器协调，限制跨多个客户端实例的 RU 消耗。
• 服务器端吞吐量控制 - 使用服务器端吞吐量存储桶限制 RU 消耗。

先决条件

• Azure Cosmos DB Java SDK v4 >= 4.13.0（用于本地和多区域吞吐量控制）
• Azure Cosmos DB Java SDK v4 >= 4.54.0（适用于变更源处理器的吞吐量控制）
• Azure Cosmos DB Java SDK v4 >= 4.74.0（用于通过吞吐量桶实现服务器端吞吐量控制）

必需的导入

本文中的代码示例使用以下导入：

import com.azure.cosmos.CosmosAsyncClient;
import com.azure.cosmos.CosmosAsyncContainer;
import com.azure.cosmos.CosmosAsyncDatabase;
import com.azure.cosmos.CosmosClientBuilder;
import com.azure.cosmos.GlobalThroughputControlConfig;
import com.azure.cosmos.ThroughputControlGroupConfig;
import com.azure.cosmos.ThroughputControlGroupConfigBuilder;
import com.azure.cosmos.models.CosmosItemRequestOptions;
import com.azure.cosmos.models.PartitionKey;
import com.azure.cosmos.models.ChangeFeedProcessor;
import com.azure.cosmos.models.ChangeFeedProcessorBuilder;
import com.azure.identity.DefaultAzureCredentialBuilder;
import com.fasterxml.jackson.databind.JsonNode;
import java.time.Duration;

创建吞吐量控制组

ThroughputControlGroupConfigBuilder用于创建吞吐量控制组配置。 可以设置：

本地吞吐量控制

本地吞吐量控制限制单个 CosmosAsyncClient 实例中的 RU 消耗量。 如果要隔离同一应用程序中不同类型的操作，这非常有用。

ThroughputControlGroupConfig groupConfig =
    new ThroughputControlGroupConfigBuilder()
        .groupName("localControlGroup")
        .targetThroughputThreshold(0.1) // limit to 10% of provisioned throughput
        .build();

container.enableLocalThroughputControlGroup(groupConfig);

启用本地吞吐量控制后，该容器上的所有操作仅限于此客户端实例中配置的 RU/s。

多个本地控件组

可以在同一容器上创建多个本地控件组。 例如，若要分隔读取和写入工作负荷，

// Limit reads to 25% of provisioned throughput
ThroughputControlGroupConfig readGroup =
    new ThroughputControlGroupConfigBuilder()
        .groupName("reads")
        .targetThroughputThreshold(0.25)
        .defaultControlGroup(true) // default group for all requests
        .build();

container.enableLocalThroughputControlGroup(readGroup);

// Limit writes to a fixed 500 RU/s
ThroughputControlGroupConfig writeGroup =
    new ThroughputControlGroupConfigBuilder()
        .groupName("writes")
        .targetThroughput(500)
        .build();

container.enableLocalThroughputControlGroup(writeGroup);

若要将特定操作分配给非默认组，请使用 CosmosItemRequestOptions：

CosmosItemRequestOptions requestOptions = new CosmosItemRequestOptions();
requestOptions.setThroughputControlGroupName("writes");

container.createItem(newItem, new PartitionKey(newItem.getPartitionKey()), requestOptions)
    .block();

全局吞吐量控制

全局吞吐量控制通过使用Azure Cosmos DB中的共享元数据容器协调多个客户端实例的 RU 消耗。 当有多个微服务或应用程序实例访问同一容器并且想要强制实施集体吞吐量限制时，这非常有用。

设置控制容器

多区域吞吐量控制功能需要元数据容器来跟踪客户端的 RU 使用情况。 可以为此创建专用数据库和容器，或使用现有容器。

// Create a dedicated database and container for throughput control metadata
client.createDatabaseIfNotExists("ThroughputControlDatabase").block();
CosmosAsyncDatabase throughputControlDatabase =
    client.getDatabase("ThroughputControlDatabase");

throughputControlDatabase
    .createContainerIfNotExists("ThroughputControlContainer", "/groupId")
    .block();

启用多区域吞吐量控制

// Create the throughput control group config
ThroughputControlGroupConfig groupConfig =
    new ThroughputControlGroupConfigBuilder()
        .groupName("globalControlGroup")
        .targetThroughputThreshold(0.25) // limit to 25% of provisioned throughput
        .defaultControlGroup(true)
        .build();

// Create the global control config pointing to the metadata container
GlobalThroughputControlConfig globalControlConfig =
    client.createGlobalThroughputControlConfigBuilder(
            "ThroughputControlDatabase",
            "ThroughputControlContainer")
        .setControlItemRenewInterval(Duration.ofSeconds(5))
        .setControlItemExpireInterval(Duration.ofSeconds(11))
        .build();

// Enable multiple-regional throughput control on the target container
container.enableGlobalThroughputControlGroup(groupConfig, globalControlConfig);

多区域吞吐量控制的配置选项

服务器端吞吐量控制

服务器端吞吐量控制使用 吞吐量存储桶 在服务器级别强制实施 RU 限制。 此模式需要Azure Cosmos DB Java SDK v4 版本 4.74.0 或更高版本。

ThroughputControlGroupConfig groupConfig =
    new ThroughputControlGroupConfigBuilder()
        .groupName("serverSideControlGroup")
        .throughputBucket(2) // assign to server-side bucket 2
        .build();

container.enableServerThroughputControlGroup(groupConfig);

有关吞吐量存储桶的详细信息，请参阅 Azure Cosmos DB 中的吞吐量存储桶。

错误处理

当请求超出为其控制组分配的 RU 预算时，SDK 会在本地拒绝该请求并返回 CosmosException 状态 429 (Too Many Requests)，而不向服务发送请求。 应用程序必须像往常一样继续处理 429 个响应（SDK 的内置重试策略适用）。

若要正常处理初始化错误，请设置为continueOnInitErrortrue：

ThroughputControlGroupConfig groupConfig =
    new ThroughputControlGroupConfigBuilder()
        .groupName("resilientGroup")
        .targetThroughputThreshold(0.5)
        .continueOnInitError(true) // fall back to full throughput on init failure
        .build();

使用更改源处理器进行吞吐量控制

可以将吞吐量控制应用于 更改源处理器 ，以限制更改源处理期间的 RU 消耗。 这在回填场景中尤为有用，尤其是在更改源处理器读取大量数据时。 此集成需要Azure Cosmos DB Java SDK v4 版本 4.54.0 或更高版本。

若要在更改源处理器上启用吞吐量控制，请在生成处理器之前在源容器上配置吞吐量控制组：

// Configure throughput control on the feed container
ThroughputControlGroupConfig groupConfig =
    new ThroughputControlGroupConfigBuilder()
        .groupName("cfpControlGroup")
        .targetThroughputThreshold(0.3) // limit to 30% of provisioned throughput
        .defaultControlGroup(true)
        .build();

feedContainer.enableLocalThroughputControlGroup(groupConfig);

// Build the Change Feed Processor — it inherits the throughput control settings
ChangeFeedProcessor changeFeedProcessor = new ChangeFeedProcessorBuilder()
    .hostName("host-1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(docs -> {
        for (JsonNode doc : docs) {
            // process each change
        }
    })
    .buildChangeFeedProcessor();

还可以将多区域吞吐量控制与更改源处理器配合使用，以跨多个处理器实例协调 RU 限制。

相关内容

• Azure Cosmos DB Spark 连接器中的吞吐量控制
• Azure Cosmos DB 中的吞吐量存储桶
• Azure Cosmos DB 中的更改源处理程序
• Azure Cosmos DB 中基于优先级的执行
• Azure Cosmos DB 中的请求单位
• Java SDK v4 最佳实践
• 适用于 Java SDK v4 的性能提示