변경 피드 추정기 사용

이 문서에서는 변경 피드를 읽을 때 변경 피드 프로세서 인스턴스의 진행률을 모니터링하는 방법을 설명합니다.

모니터링 진행률이 중요한 이유는 무엇인가요?

변경 피드 프로세서는 변경 피드 전체에서 앞으로 이동하고 변경 내용을 대리자 구현에 전달하는 포인터 역할을 합니다.

변경 피드 프로세서 배포는 CPU, 메모리, 네트워크 등과 같은 사용 가능한 리소스에 따라 특정 속도로 변경 내용을 처리할 수 있습니다.

이 속도가 Azure Cosmos DB 컨테이너에서 변경이 발생하는 속도보다 느리면 프로세서가 뒤처지기 시작합니다.

이 시나리오를 식별하면 변경 피드 프로세서 배포의 크기를 조정해야 하는지 이해하는 데 도움이 됩니다.

변경 피드 예측 도구 구현

.NET

자동 알림에 대한 푸시 모델

변경 피드 프로세서와 마찬가지로 변경 피드 추정기는 푸시 모델로 작동할 수 있습니다. 예측 도구는 마지막으로 처리된 항목(임대 컨테이너의 상태로 정의됨)과 컨테이너의 최신 변경 내용 간의 차이를 측정하고 이 값을 대리자에 푸시합니다. 측정을 수행할 간격을 기본값인 5초로 사용자 지정할 수도 있습니다.

예를 들어 변경 피드 프로세서가 최신 버전 모드를 사용하고 다음과 같이 정의된 경우:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

프로세서를 측정하기 위해 추정기를 초기화하는 올바른 방법은 GetChangeFeedEstimatorBuilder를 사용하는 것입니다.

ChangeFeedProcessor changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimatorBuilder("changeFeedEstimator", Program.HandleEstimationAsync, TimeSpan.FromMilliseconds(1000))
    .WithLeaseContainer(leaseContainer)
    .Build();

프로세서와 추정기가 동일한 leaseContainer와 같은 이름을 공유하는 위치입니다.

다른 두 매개 변수는 프로세서에서 읽을 보류 중인 변경 내용 수 와 이 측정을 수행할 시간 간격을 나타내는 숫자를 받는 대리자입니다.

추정을 받는 대리자의 예는 다음과 같습니다.

static async Task HandleEstimationAsync(long estimation, CancellationToken cancellationToken)
{
    if (estimation > 0)
    {
        Console.WriteLine($"\tEstimator detected {estimation} items pending to be read by the Processor.");
    }

    await Task.Delay(0);
}

이 추정을 모니터링 솔루션에 보내고 이를 사용하여 시간이 지남에 따라 진행률이 어떻게 작동하는지 이해할 수 있습니다.

주문형 상세 예측으로

푸시 모델과 달리 요청 시 예상을 얻을 수 있는 대안이 있습니다. 이 모델은 더 자세한 정보도 제공합니다.

• 임대당 예상 지연 시간입니다.
• 인스턴스에서 각 임대를 소유하고 처리하므로 인스턴스에 문제가 있는지 확인할 수 있습니다.

변경 피드 프로세서가 다음과 같이 정의된 경우:

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("changeFeedEstimator", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .Build();

동일한 임대 구성으로 추정기를 만들 수 있습니다.

ChangeFeedEstimator changeFeedEstimator = monitoredContainer
    .GetChangeFeedEstimator("changeFeedEstimator", leaseContainer);

필요할 때마다 필요한 빈도를 사용하여 자세한 추정을 얻을 수 있습니다.

Console.WriteLine("Checking estimation...");
using FeedIterator<ChangeFeedProcessorState> estimatorIterator = changeFeedEstimator.GetCurrentStateIterator();
while (estimatorIterator.HasMoreResults)
{
    FeedResponse<ChangeFeedProcessorState> states = await estimatorIterator.ReadNextAsync();
    foreach (ChangeFeedProcessorState leaseState in states)
    {
        string host = leaseState.InstanceName == null ? $"not owned by any host currently" : $"owned by host {leaseState.InstanceName}";
        Console.WriteLine($"Lease [{leaseState.LeaseToken}] {host} reports {leaseState.EstimatedLag} as estimated lag.");
    }
}

각각의 ChangeFeedProcessorState에는 임대 및 지연 정보가 포함되어 있으며, 이를 소유하고 있는 현재 인스턴스도 포함됩니다.

예측 도구 배포

변경 피드 예측 도구는 변경 피드 프로세서의 일부로 배포하거나 동일한 프로젝트에 속할 필요가 없습니다. 프로세서에서 독립 인스턴스에 추정기를 배포하는 것이 좋습니다. 단일 추정기 인스턴스는 변경 피드 프로세서 배포의 모든 임대 및 인스턴스에 대한 진행률을 추적할 수 있습니다.

각 예측은 모니터링된 컨테이너 및 임대 컨테이너의 요청 단위를 사용합니다. 1분 사이의 빈도는 좋은 시작점이며, 빈도가 낮을수록 소비된 요청 단위가 높아집니다.

java

이 예제에서는 변경 피드 이벤트를 처리하는 데 지연이 예상되는 최신 버전 모드의 변경 피드 프로세서를 사용합니다. 애플리케이션에서 문서는 하나의 컨테이너("피드 컨테이너")에 삽입되고 다른 작업자 스레드 또는 작업자 애플리케이션은 피드 컨테이너의 변경 피드에서 삽입된 문서를 끌어와서 어떤 식으로든 작동합니다.

변경 피드 프로세서는 다음과 같이 빌드되고 시작됩니다.

ChangeFeedProcessor changeFeedProcessorMainInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChangesWithLag())
    .buildChangeFeedProcessor();

try {
    changeFeedProcessorMainInstance
        .start()
        .subscribeOn(Schedulers.boundedElastic())
        .timeout(Duration.ofSeconds(10))
        .then(Mono.just(changeFeedProcessorMainInstance)
                  .delayElement(Duration.ofSeconds(10))
                  .flatMap(value -> changeFeedProcessorMainInstance
                      .stop()
                      .subscribeOn(Schedulers.boundedElastic())
                      .timeout(Duration.ofSeconds(10))))
        .subscribe();
} catch (Exception ex) {
    logger.error("Change feed processor did not start and stopped in the expected time", ex);
    throw ex;
}

동일한 입력 컨테이너(feedContainer 및 leaseContainer)와 정확히 동일한 임대 접두사()가 사용되는 한, 변경 피드 프로세서 지연 검사는 별도의 애플리케이션(CONTAINER_NAME + "-lease"예: 상태 모니터)에서 수행할 수 있습니다. 추정기 코드를 사용하려면 변경 피드 프로세서가 leaseContainer의 문서를 완전히 초기화할 수 있어야 합니다.

예측 도구는 변경 피드 프로세서의 현재 상태를 검색하고 처리 중인 각 이벤트에 대한 예상 지연 시간 값을 합산하여 누적 지연을 계산합니다.

Mono<List<ChangeFeedProcessorState>> currentState = changeFeedProcessorMainInstance.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

totalLag.set(0);
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).subscribe();

// Initially totalLag should be zero
logger.info("Initially total lag is : {}", totalLag.get());

처음에는 총 지연 시간이 0이어야 하며 마지막으로 만든 문서 수보다 크거나 같아야 합니다. 총 지연 시간 값을 기록하거나 추가 분석에 사용할 수 있으므로 변경 피드 처리의 성능을 모니터링하고 시스템의 잠재적인 병목 상태 또는 지연을 식별할 수 있습니다.

totalLag.set(0);
changeFeedProcessorSideCart.start().block();
currentState = changeFeedProcessorSideCart.getCurrentState();
currentState.map(state -> {
    for (ChangeFeedProcessorState changeFeedProcessorState : state) {
        totalLag.addAndGet(changeFeedProcessorState.getEstimatedLag());
    }
    return state;
}).block();

// Finally, totalLag should be greater or equal to the number of documents created
logger.info("Finally total lag is : {}", totalLag.get());

변경 내용을 수신하고 지연으로 처리하는 대리자의 예는 다음과 같습니다.

private static Consumer<List<JsonNode>> handleChangesWithLag() {
    return (List<JsonNode> docs) -> {
        logger.info("Start handleChangesWithLag()");

        try {
            Thread.sleep(Duration.ofSeconds(5).toMillis());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        for (JsonNode document : docs) {
            try {
                //Change Feed hands the document to you in the form of a JsonNode
                //As a developer you have two options for handling the JsonNode document provided to you by Change Feed
                //One option is to operate on the document in the form of a JsonNode, as shown below. This is great
                //especially if you do not have a single uniform data model for all documents.
                logger.info("Document received: " + OBJECT_MAPPER.writerWithDefaultPrettyPrinter()
                                                                 .writeValueAsString(document));

                //You can also transform the JsonNode to a POJO having the same structure as the JsonNode,
                //as shown below. Then you can operate on the POJO.
                CustomPOJO2 pojo_doc = OBJECT_MAPPER.treeToValue(document, CustomPOJO2.class);
                logger.info("id: " + pojo_doc.getId());

            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }
        logger.info("End handleChangesWithLag()");
    };
}

지원되는 변경 피드 모드

변경 피드 추정기는 최신 버전 모드 와 모든 버전 및 삭제 모드 모두에 사용할 수 있습니다. 두 모드에서 제공된 예상 값이 처리할 미해결 변경 내용의 정확한 개수로 보장되지는 않습니다.

추가 리소스

• Azure Cosmos DB SDK
• GitHub의 사용 샘플(.NET 최신 버전)
• GitHub의 사용 샘플(.NET 모든 버전 및 삭제)
• GitHub의 사용 샘플(Java)
• GitHub의 추가 샘플

다음 단계

이제 다음 문서에서 변경 피드 프로세서에 대해 자세히 알아볼 수 있습니다.

• 변경 피드 프로세서 개요