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

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.

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.

Kontekst

Ź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 zaawansowana, aby w razie potrzeby dystrybuować zdarzenia zestawienia zmian między wieloma procesami roboczymi. Wystarczy przekazać bibliotekę zestawienia 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 danych", aby naśladować przepływ danych. Procesor rozpływu zmian, powiązany z kontenerem źródłowym, 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/

Walkthrough

1. Skonfiguruj ChangeFeedProcessorOptions w aplikacji Java, używając Azure Cosmos DB i Java SDK Azure Cosmos DB V4. ChangeFeedProcessorOptions zapewnia podstawowe ustawienia umożliwiające kontrolowanie zachowania Change Feed Processor podczas przetwarzania danych.

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

2. Zainicjuj ChangeFeedProcessor z odpowiednimi konfiguracjami, w tym z nazwą hosta, kontenerem źródłowym, kontenerem na dzierżawy i logiką obsługi danych. Metoda start() inicjuje przetwarzanie danych, umożliwiając współbieżne i w czasie rzeczywistym przetwarzanie przychodzących zmian danych z kontenera źródłowego.

   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 delegata, który obsługuje zmiany danych przekazywanych za pomocą metody handleChanges(). Metoda przetwarza odebrane dokumenty JsonNode z Change Feed. Jako deweloper masz dwie opcje obsługi dokumentu JsonNode dostarczonego przez Strumień zmian. Jedną z opcji jest operowanie na dokumencie w postaci węzła JSON. 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 Zmiennych Strumieni, wstawia przykładowe dokumenty do kontenera strumienia i przetwarza nadchodzą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

• Azure Cosmos DB Java SDK V4
• Dodatkowe przykłady w witrynie GitHub

Dalsze kroki

Teraz możesz dowiedzieć się więcej na temat estymatora strumienia zmian w następujących artykułach:

• Użyj oszacowania zmiany strumienia