NoSQL için Azure Cosmos DB kullanan bir Java uygulaması oluşturma ve akış işlemcisini değiştirme

  • Makale

UYGULANANLAR: NOSQL

Azure Cosmos DB, Microsoft tarafından sağlanan tam olarak yönetilen bir NoSQL veritabanı hizmetidir. Küresel olarak dağıtılmış ve yüksek oranda ölçeklenebilir uygulamaları kolayca oluşturmanıza olanak tanır. Bu nasıl yapılır kılavuzu, NoSQL için Azure Cosmos DB veritabanını kullanan ve gerçek zamanlı veri işleme için Değişiklik Akışı İşlemcisi'ni uygulayan bir Java uygulaması oluşturma işleminde size yol gösterir. Java uygulaması, Azure Cosmos DB Java SDK v4 kullanarak NoSQL için Azure Cosmos DB ile iletişim kurar.

Önemli

Bu öğretici yalnızca Azure Cosmos DB Java SDK v4 içindir. Daha fazla bilgi için lütfen Azure Cosmos DB Java SDK v4 Sürüm notları, Maven deposu, Azure Cosmos DB'de değişiklik akışı işlemcisi ve Azure Cosmos DB Java SDK v4 sorun giderme kılavuzunu görüntüleyin. Şu anda v4'ten daha eski bir sürüm kullanıyorsanız v4'e yükseltme konusunda yardım için Bkz. Azure Cosmos DB Java SDK'sına geçiş v4 kılavuzu.

Önkoşullar

Arka Plan

Azure Cosmos DB değişiklik akışı, birçok kullanımı olan belge eklemeye yanıt olarak eylemleri tetikleyen olay temelli bir arabirim sağlar.

Değişiklik akışı olaylarını yönetme çalışmalarına büyük ölçüde SDK'da yerleşik olarak bulunan değişiklik akışı İşlemci kitaplığı dikkate alınır. Bu kitaplık, değişiklik akışı olaylarını isteniyorsa birden çok çalışan arasında dağıtacak kadar güçlüdür. Tek yapmanız gereken değişiklik akışı kitaplığına bir geri çağırma sağlamaktır.

Java uygulamasının bu basit örneği, Azure Cosmos DB ve Değişiklik Akışı İşlemcisi ile gerçek zamanlı veri işlemeyi göstermektir. Uygulama, veri akışının benzetimini yapmak için örnek belgeleri "akış kapsayıcısına" ekler. Akış kapsayıcısına bağlı Değişiklik Akışı İşlemcisi, gelen değişiklikleri işler ve belge içeriğini günlüğe kaydeder. İşlemci paralel işleme için kiraları otomatik olarak yönetir.

Kaynak kod

SDK örnek depoyu kopyalayabilir ve bu örneği içinde SampleChangeFeedProcessor.javabulabilirsiniz:

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/

Kılavuz

  1. ChangeFeedProcessorOptions Azure Cosmos DB ve Azure Cosmos DB Java SDK V4 kullanarak bir Java uygulamasında öğesini yapılandırın. veri ChangeFeedProcessorOptions işleme sırasında Değişiklik Akışı İşlemcisi'nin davranışını denetlemek için gerekli ayarları sağlar.

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

  2. Ana bilgisayar adı, akış kapsayıcısı, kira kapsayıcısı ve veri işleme mantığı gibi ilgili yapılandırmalarla ChangeFeedProcessor'ı başlatın. start() yöntemi veri işlemeyi başlatır ve akış kapsayıcısından gelen veri değişikliklerinin eşzamanlı ve gerçek zamanlı olarak işlenmesini sağlar.

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

  3. Temsilcinin, yöntemini kullanarak gelen veri değişikliklerini işleyeceğini handleChanges() belirtin. yöntemi, alınan JsonNode belgelerini Değişiklik Akışı'ndan işler. Geliştirici olarak, Değişiklik Akışı tarafından sağlanan JsonNode belgesini işlemek için iki seçeneğiniz vardır. Bir seçenek, belge üzerinde JsonNode biçiminde çalışmaktır. Bu, özellikle tüm belgeler için tek bir tekdüzen veri modeliniz yoksa harikadır. İkinci seçenek: JsonNode'yi JsonNode ile aynı yapıya sahip bir POJO'ya dönüştürün. Ardından POJO üzerinde çalışabilirsiniz.

    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. Java uygulamasını derleyin ve çalıştırın. Uygulama, Değişiklik Akışı İşlemcisi'ni başlatır, akış kapsayıcısına örnek belgeler ekler ve gelen değişiklikleri işler.

Sonuç

Bu kılavuzda, NoSQL için Azure Cosmos DB veritabanını kullanan ve gerçek zamanlı veri işleme için Değişiklik Akışı İşlemcisi'ni kullanan Azure Cosmos DB Java SDK V4 kullanarak bir Java uygulaması oluşturmayı öğrendiniz. Azure Cosmos DB kullanarak daha karmaşık kullanım örneklerini işlemek ve sağlam, ölçeklenebilir ve genel olarak dağıtılmış uygulamalar oluşturmak için bu uygulamayı genişletebilirsiniz.

Ek kaynaklar

Sonraki adımlar

Şimdi aşağıdaki makalelerde değişiklik akışı tahmin aracı hakkında daha fazla bilgi edinmek için devam edebilirsiniz: