Udostępnij za pomocą

Jak utworzyć aplikację Java korzystającą z usługi Azure Cosmos DB dla NoSQL i procesora strumienia zmian

DOTYCZY: NoSQL

Azure Cosmos DB to w pełni zarządzana usługa bazy danych NoSQL udostępniana przez firmę Microsoft. Umożliwia łatwe tworzenie globalnie rozproszonych i wysoce skalowalnych aplikacji. Ten przewodnik z instrukcjami przeprowadzi Cię przez proces tworzenia aplikacji Java korzystającej z bazy danych Azure Cosmos DB for NoSQL i implementuje procesor zestawienia zmian na potrzeby przetwarzania danych w czasie rzeczywistym. Aplikacja Java komunikuje się z usługą Azure Cosmos DB for NoSQL przy użyciu zestawu Java SDK usługi Azure Cosmos DB w wersji 4.

Ważne

Ten samouczek dotyczy tylko zestawu Java SDK usługi Azure Cosmos DB w wersji 4. Aby uzyskać więcej informacji, zobacz informacje o wydaniu dla Java SDK w wersji 4 usługi Azure Cosmos DB, repozytorium Maven, procesor zasobu zmian dla bazy danych Azure Cosmos oraz przewodnik rozwiązywania problemów z Java SDK w wersji 4 dla usługi Azure Cosmos DB. Jeśli obecnie używasz starszej wersji niż 4, zobacz przewodnik Migrowanie do zestawu Java SDK usługi Azure Cosmos DB w wersji 4 , aby uzyskać pomoc dotyczącą uaktualniania do wersji 4.

Wymagania wstępne

  • Konto usługi Azure Cosmos DB: możesz utworzyć je w witrynie Azure Portal lub użyć emulatora usługi Azure Cosmos DB.

  • Java Development Environment: upewnij się, że na maszynie zainstalowano zestaw Java Development Kit (JDK) z co najmniej 8 wersjami.

  • Zestaw Java SDK usługi Azure Cosmos DB w wersji 4: udostępnia niezbędne funkcje do interakcji z usługą Azure Cosmos DB.

Tło

Źródło zmian usługi Azure Cosmos DB udostępnia interfejs sterowany zdarzeniami umożliwiający wyzwalanie akcji w odpowiedzi na wstawienie dokumentu, które ma wiele zastosowań.

Praca zarządzania zdarzeniami zestawienia zmian jest w dużej mierze zajęta przez bibliotekę procesora zestawienia zmian wbudowaną w zestaw SDK. Ta biblioteka jest wystarczająco potężna, aby w razie potrzeby dystrybuować zdarzenia strumienia zmian między wieloma pracownikami. Wystarczy, że dostarczysz bibliotece zmian wywołanie zwrotne.

Ten prosty przykład aplikacji Java pokazuje przetwarzanie danych w czasie rzeczywistym za pomocą usługi Azure Cosmos DB i procesora zestawienia zmian. Aplikacja wstawia przykładowe dokumenty do "kontenera", aby symulować strumień danych. Procesor kanału zmian, związany z kontenerem kanału danych, przetwarza przychodzące zmiany i rejestruje zawartość dokumentu. Procesor automatycznie zarządza dzierżawami przetwarzania równoległego.

Kod źródłowy

Możesz sklonować przykładowe repozytorium zestawu SDK i znaleźć ten przykład w pliku SampleChangeFeedProcessor.java:

git clone https://github.com/Azure-Samples/azure-cosmos-java-sql-api-samples.git
cd azure-cosmos-java-sql-api-sample/src/main/java/com/azure/cosmos/examples/changefeed/

Przewodnik krok po kroku

  1. Skonfiguruj ChangeFeedProcessorOptions w aplikacji Java przy użyciu Azure Cosmos DB i Azure Cosmos DB Java SDK V4. Zapewnia ChangeFeedProcessorOptions podstawowe ustawienia umożliwiające kontrolowanie zachowania procesora zestawienia zmian podczas przetwarzania danych.

    options = new ChangeFeedProcessorOptions();
options.setStartFromBeginning(false);
options.setLeasePrefix("myChangeFeedDeploymentUnit");
options.setFeedPollDelay(Duration.ofSeconds(5));
options.setFeedPollThroughputControlConfig(throughputControlGroupConfig);

  2. Zainicjuj ChangeFeedProcessor przy użyciu odpowiednich konfiguracji, w tym nazwy hosta, kontenera danych strumieniowych, kontenera dzierżawy i logiki przetwarzania danych. Metoda start() inicjuje przetwarzanie danych, umożliwiając jednoczesne i w czasie rzeczywistym przetwarzanie przychodzących zmian danych z pojemnika na dane.

    logger.info("Start Change Feed Processor on worker (handles changes asynchronously)");
ChangeFeedProcessor changeFeedProcessorInstance = new ChangeFeedProcessorBuilder()
    .hostName("SampleHost_1")
    .feedContainer(feedContainer)
    .leaseContainer(leaseContainer)
    .handleChanges(handleChanges())
    .options(options)
    .buildChangeFeedProcessor();
changeFeedProcessorInstance.start()
                           .subscribeOn(Schedulers.boundedElastic())
                           .subscribe();

  3. Określ delegat obsługuje zmiany danych przychodzących przy użyciu handleChanges() metody . Metoda przetwarza odebrane dokumenty JsonNode z zestawienia zmian. Jako deweloper masz dwie opcje obsługi dokumentu JsonNode dostarczonego przez zestawienie zmian. Jedną z opcji jest obsługa dokumentu w postaci węzła JsonNode. Jest to doskonałe rozwiązanie, zwłaszcza jeśli nie masz jednego jednolitego modelu danych dla wszystkich dokumentów. Druga opcja — przekształcenie węzła JsonNode w obiekt POJO o tej samej strukturze co węzeł JsonNode. Następnie można pracować na poJO.

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

        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();
            }
        }
        isWorkCompleted = true;
        logger.info("End handleChanges()");

    };
}

  4. Skompiluj i uruchom aplikację Java. Aplikacja uruchamia procesor zestawienia zmian, wstawia przykładowe dokumenty do kontenera kanału informacyjnego i przetwarza przychodzące zmiany.

Podsumowanie

W tym przewodniku przedstawiono sposób tworzenia aplikacji Java przy użyciu zestawu Java SDK usługi Azure Cosmos DB w wersji 4 korzystającej z bazy danych Azure Cosmos DB for NoSQL i używania procesora zestawienia zmian do przetwarzania danych w czasie rzeczywistym. Tę aplikację można rozszerzyć, aby obsługiwać bardziej złożone przypadki użycia i tworzyć niezawodne, skalowalne i globalnie rozproszone aplikacje przy użyciu usługi Azure Cosmos DB.

Dodatkowe zasoby

Następne kroki

Możesz dowiedzieć się więcej o estymatorze strumienia zmian w następujących artykułach:

  • Last updated on