Spring Tümleştirmesi için Spring Cloud Azure desteği

  • Şunlara uygulanır: ✅ Version 4.20.0, ✅ Version 5.25.0, ✅ Version 6.5.0, ✅ Version 7.4.0

Azure için Spring Tümleştirme Uzantısı, Javaiçin Azure SDK tarafından sağlanan çeşitli hizmetler için Spring Tümleştirme bağdaştırıcıları sağlar. Şu Azure hizmetleri için Spring Integration desteği sağlıyoruz: Event Hubs, Service Bus, Depolama Kuyruğu. Desteklenen bağdaştırıcıların listesi aşağıdadır:

  • spring-cloud-azure-starter-integration-eventhubs - Daha fazla bilgi için bkz. Azure Event Hubs ile Spring Tümleştirmesi
  • - Daha fazla bilgi için bkz. Azure Service Bus ile Spring Integration
  • spring-cloud-azure-starter-integration-storage-queue - Daha fazla bilgi için bkz. Azure Depolama Kuyruğu ile Spring Tümleştirmesi

Azure Event Hubs ile Spring Tümleştirmesi

Temel kavramlar

Azure Event Hubs, büyük bir veri akışı platformu ve olay alımı hizmetidir. Saniyede milyonlarca olay alabilir ve işleyebilir. Bir olay hub'ına gönderilen veriler, herhangi bir gerçek zamanlı analiz sağlayıcısı veya toplu işlem/depolama bağdaştırıcısı kullanılarak dönüştürülebilir ve depolanabilir.

Spring Integration, Spring tabanlı uygulamalarda basit mesajlaşmaya olanak tanır ve bildirim temelli bağdaştırıcılar aracılığıyla dış sistemlerle tümleştirmeyi destekler. Bu bağdaştırıcılar Spring'in uzaktan iletişim, mesajlaşma ve zamanlama desteği konusunda daha yüksek düzeyde soyutlama sağlar. Event Hubs için Spring Integration uzantısı projesi, Azure Event Hubs için gelen ve giden kanal bağdaştırıcıları ve ağ geçitleri sağlar.

Not

RxJava destek API'leri 4.0.0 sürümünden bırakılır. Ayrıntılar için bkz. Javadoc.

Tüketici grubu

Event Hubs, tüketici grubu için Apache Kafka ile benzer destek sağlar ancak biraz farklı mantık sunar. Kafka, kaydedilen tüm ofsetleri broker'da depolarken, işlenmekte olan Event Hubs iletilerinin ofsetlerini manuel olarak depolamanız gerekir. Event Hubs SDK'sı, bu tür uzaklıkları Azure Depolama'da depolama işlevi sağlar.

Bölümleme desteği

Event Hubs, Kafka ile benzer bir fiziksel bölüm kavramı sağlar. Ancak Kafka'nın tüketiciler ve bölümler arasında otomatik yeniden dengelemeden farklı olarak Event Hubs bir tür önleyici mod sağlar. Depolama hesabı, hangi bölümün hangi tüketiciye ait olduğunu belirlemek için kira görevi görür. Yeni bir tüketici başladığında, iş yükü dengelemesini sağlamak için en ağır yüklü tüketicilerden bazı bölümleri devralmaya çalışır.

Geliştiriciler, yük dengeleme stratejisini belirtmek için yapılandırma için EventHubsContainerProperties kullanabilir. nasıl yapılandırılacağına dair bir örnek için aşağıdaki bölüme bakın.

Toplu tüketici desteği

EventHubsInboundChannelAdapter toplu tüketim modunu destekler. Kullanıcılar bunu etkinleştirmek için ListenerMode.BATCH örneği oluştururken dinleyici modunu EventHubsInboundChannelAdapter olarak belirtebilir. Etkinleştirildiğinde, yükü toplu hâle getirilmiş olayların bir listesi olan bir Message alınacak ve alt kanala iletilecektir. Her mesaj üst bilgisi de, içeriği her olaydan ayrıştırılan ilgili başlık değeri olan bir listeye dönüştürülür. Bölüm kimliği, denetim noktası oluşturucu ve son sıraya alınan özelliklerin ortak üst bilgileri için, olay grubunun tamamı için tek bir değer olarak sunulurlar ve aynı değeri paylaşırlar. Daha fazla bilgi için Event Hubs İleti Üst Bilgileri bölümüne bakın.

Not

Denetim noktası üst bilgisi yalnızca MANUAL denetim noktası modu kullanıldığında bulunur.

Toplu tüketici için denetim noktası oluşturma iki modu destekler: BATCH ve MANUAL. BATCH modu, olayların tamamı alındıktan sonra tüm olay grubuna tek seferde denetim noktası oluşturan otomatik bir denetim noktası modudur. MANUAL modu, kullanıcılara göre olayları kontrol etmektir. Kullanıldığında, Checkpointer ileti üst bilgisine eklenir ve kullanıcılar bunu denetim noktası alma işlemi için kullanabilir.

Toplu tüketim ilkesi, max-size ve max-wait-time özellikleriyle belirtilebilir; burada max-size gerekli bir özelliktir, max-wait-time ise isteğe bağlı bir özelliktir. Geliştiriciler toplu kullanım stratejisini belirtmek için yapılandırma için EventHubsContainerProperties kullanabilir. 'in nasıl yapılandırılacağına ilişkin bir örnek için aşağıdaki bölüme bakın.

Bağımlılık kurulumu

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-eventhubs</artifactId>
</dependency>

Konfigürasyon

Bu başlatıcı, yapılandırma seçeneklerinin aşağıdaki 3 bölümünü sağlar:

Bağlantı Yapılandırma Özellikleri

Bu bölüm, Azure Event Hubs'a bağlanmak için kullanılan yapılandırma seçeneklerini içerir.

Not

Azure kaynağına erişmek için Microsoft Entra ID ile kimlik doğrulaması yapmak ve yetkilendirme sağlamak üzere bir güvenlik sorumlusu kullanmayı seçerseniz, güvenlik sorumlusuna Azure kaynağına erişmesi için yeterli izinlerin verildiğinden emin olmak üzere Microsoft Entra ID ile erişimi yetkilendirme bölümüne bakın.

spring-cloud-azure-starter-integration-eventhubs bağlantısının yapılandırılabilir özellikleri:

Mülk Tip Açıklama
spring.cloud.azure.eventhubs.enabled Boolean Azure Event Hubs'ın etkinleştirilip etkinleştirilmediği.
spring.cloud.azure.eventhubs.connection-string Dize Event Hubs Ad Alanı bağlantı dizesi değeri.
spring.cloud.azure.eventhubs.namespace Dize FQDN’nin ön eki olan Event Hubs ad alanı değeri. FQDN, NamespaceName.DomainName dosyasından oluşmalıdır
spring.cloud.azure.eventhubs.domain-name Dize Bir Azure Event Hubs Ad Alanı değerinin etki alanı adı.
spring.cloud.azure.eventhubs.custom-endpoint-address Dize Özel Uç Nokta adresi.
spring.cloud.azure.eventhubs.shared-connection Boolean (Boole Mantığı) Temeldeki EventProcessorClient ve EventHubProducerAsyncClient'ın aynı bağlantıyı kullanıp kullanmadığı. Varsayılan olarak, oluşturulan her Event Hub istemcisi için yeni bir bağlantı oluşturulur ve kullanılır.

Denetim Noktası Yapılandırma Özellikleri

Bu bölüm, bölüm sahipliğini ve denetim noktası bilgilerini kalıcı hale etmek için kullanılan Depolama Blobları hizmetinin yapılandırma seçeneklerini içerir.

Not

4.0.0 sürümünden spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists özelliği el ile etkinleştirilmediğinde depolama kapsayıcısı otomatik olarak oluşturulmaz.

spring-cloud-azure-starter-integration-eventhubs öğesinin yapılandırılabilir özelliklerinin denetlenmesi:

Mülk Tip Açıklama
spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists Boolean (Boole Mantığı) Mevcut değilse kapsayıcı oluşturmaya izin verilip verilmeyeceği.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name Dize Depolama hesabının adı.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key Dize Depolama hesabı erişim anahtarı.
spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name Dize Depolama kapsayıcısı adı.

Yaygın Azure Hizmeti SDK yapılandırma seçenekleri, Depolama Blobu denetim noktası deposu için de yapılandırılabilir. Desteklenen yapılandırma seçenekleri Spring Cloud Azure yapılandırmasında tanıtılmaktadır ve hem birleşik spring.cloud.azure. ön ekiyle hem de spring.cloud.azure.eventhubs.processor.checkpoint-store. ön ekiyle yapılandırılabilir.

Event Hub işlemci yapılandırma özellikleri

EventHubsInboundChannelAdapter, bir olay hub’ından iletileri almak için EventProcessorClient kullanır; bir EventProcessorClient’nin genel özelliklerini yapılandırmak için geliştiriciler yapılandırma amacıyla EventHubsContainerProperties kullanabilir. ile nasıl çalışılacağına dair aşağıdaki bölüme bakın.

Temel kullanım

Azure Event Hubs'a ileti gönderme

  1. Kimlik bilgisi yapılandırma seçeneklerini doldurun.

    • Bağlantı dizesi olarak kimlik bilgileri için application.yml dosyanızda aşağıdaki özellikleri yapılandırın:

      spring:
  cloud:
    azure:
      eventhubs:
        connection-string: ${AZURE_EVENT_HUBS_CONNECTION_STRING}
        processor:
          checkpoint-store:
            container-name: ${CHECKPOINT-CONTAINER}
            account-name: ${CHECKPOINT-STORAGE-ACCOUNT}
            account-key: ${CHECKPOINT-ACCESS-KEY}

      Not

      Microsoft, kullanılabilir en güvenli kimlik doğrulama akışının kullanılmasını önerir. Veritabanları, önbellekler, mesajlaşma veya yapay zeka hizmetleri gibi bu yordamda açıklanan kimlik doğrulama akışı, uygulamaya çok yüksek düzeyde güven gerektirir ve diğer akışlarda mevcut olmayan riskler taşır. Bu akışı yalnızca parolasız veya anahtarsız bağlantılar için yönetilen kimlikler gibi daha güvenli seçenekler uygun olmadığında kullanın. Yerel makine işlemleri için parolasız veya anahtarsız bağlantılar için kullanıcı kimliklerini tercih edin.

    • Yönetilen kimlik olarak kimlik bilgileri için application.yml dosyanızda aşağıdaki özellikleri yapılandırın:

      spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_CLIENT_ID}
      eventhubs:
        namespace: ${AZURE_EVENT_HUBS_NAMESPACE}
        processor:
          checkpoint-store:
            container-name: ${CONTAINER_NAME}
            account-name: ${ACCOUNT_NAME}

    • Hizmet sorumlusu olarak kimlik bilgileri için application.yml dosyanızda aşağıdaki özellikleri yapılandırın:

      spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      eventhubs:
        namespace: ${AZURE_EVENT_HUBS_NAMESPACE}
        processor:
          checkpoint-store:
            container-name: ${CONTAINER_NAME}
            account-name: ${ACCOUNT_NAME}

Not

tenant-id için izin verilen değerler şunlardır: common, organizations, consumersveya kiracı kimliği. Bu değerler hakkında daha fazla bilgi için, Hata AADSTS50020 - Kimlik sağlayıcısındaki kullanıcı hesabı kiracıda mevcut değil başlıklı konunun Yanlış uç nokta kullanıldı (kişisel ve kuruluş hesapları) bölümüne bakın. Tek kiracılı uygulamanızı dönüştürme hakkında bilgi için bkz. Tek kiracılı uygulamayı Microsoft Entra IDüzerinde çok kiracılıya dönüştürme.

  1. Event Hubs'a ileti göndermek için DefaultMessageHandler çekirdeğiyle EventHubsTemplate oluşturun.

    class Demo {
    private static final String OUTPUT_CHANNEL = "output";
    private static final String EVENTHUB_NAME = "eh1";

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public MessageHandler messageSender(EventHubsTemplate eventHubsTemplate) {
        DefaultMessageHandler handler = new DefaultMessageHandler(EVENTHUB_NAME, eventHubsTemplate);
        handler.setSendCallback(new ListenableFutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                LOGGER.info("Message was sent successfully.");
            }
            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error("There was an error sending the message.", ex);
            }
        });
        return handler;
    }
}

  2. Bir ileti kanalı aracılığıyla yukarıdaki ileti işleyicisiyle bir ileti ağ geçidi bağlaması oluşturun.

    class Demo {
    @Autowired
    EventHubOutboundGateway messagingGateway;

    @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
    public interface EventHubOutboundGateway {
        void send(String text);
    }
}

  3. Ağ geçidini kullanarak ileti gönderme.

    class Demo {
    public void demo() {
        this.messagingGateway.send(message);
    }
}

Azure Event Hubs'dan ileti alma

  1. Kimlik bilgisi yapılandırma seçeneklerini doldurun.

  2. Giriş kanalı olarak bir ileti kanalı çekirdeği oluşturun.

    @Configuration
class Demo {
    @Bean
    public MessageChannel input() {
        return new DirectChannel();
    }
}

  3. Event Hubs'dan mesaj almak için EventHubsMessageListenerContainer bean'i ile EventHubsInboundChannelAdapter oluşturun.

    @Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";
    private static final String EVENTHUB_NAME = "eh1";
    private static final String CONSUMER_GROUP = "$Default";

    @Bean
    public EventHubsInboundChannelAdapter messageChannelAdapter(
            @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
            EventHubsMessageListenerContainer listenerContainer) {
        EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(listenerContainer);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }

    @Bean
    public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
        EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
        containerProperties.setEventHubName(EVENTHUB_NAME);
        containerProperties.setConsumerGroup(CONSUMER_GROUP);
        containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
        return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
    }
}

  4. Daha önce oluşturulan ileti kanalı aracılığıyla EventHubsInboundChannelAdapter ile bir ileti alıcısı bağlaması oluşturun.

    class Demo {
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
        String message = new String(payload);
        LOGGER.info("New message received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
                .doOnError(e -> LOGGER.error("Error found", e))
                .block();
    }
}

ObjectMapper'ı özelleştirmek için EventHubsMessageConverter'ı yapılandırma

EventHubsMessageConverter, kullanıcıların ObjectMapper'ı özelleştirmesine izin vermek için yapılandırılabilir bir fasulye olarak yapılır.

Toplu tüketici desteği

Event Hubs'tan gelen iletileri toplu olarak kullanmak yukarıdaki örneğe benzerdir; ayrıca, kullanıcıların EventHubsInboundChannelAdapteriçin toplu işlem kullanan ilgili yapılandırma seçeneklerini ayarlaması gerekir.

EventHubsInboundChannelAdapteroluştururken dinleyici modu BATCHolarak ayarlanmalıdır. EventHubsMessageListenerContainerçekirdeği oluşturduğunuzda, denetim noktası modunu MANUAL veya BATCHolarak ayarlayın; toplu iş seçenekleri gerektiği gibi yapılandırılabilir.

@Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";
    private static final String EVENTHUB_NAME = "eh1";
    private static final String CONSUMER_GROUP = "$Default";

    @Bean
    public EventHubsInboundChannelAdapter messageChannelAdapter(
            @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
            EventHubsMessageListenerContainer listenerContainer) {
        EventHubsInboundChannelAdapter adapter = new EventHubsInboundChannelAdapter(listenerContainer, ListenerMode.BATCH);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }

    @Bean
    public EventHubsMessageListenerContainer messageListenerContainer(EventHubsProcessorFactory processorFactory) {
        EventHubsContainerProperties containerProperties = new EventHubsContainerProperties();
        containerProperties.setEventHubName(EVENTHUB_NAME);
        containerProperties.setConsumerGroup(CONSUMER_GROUP);
        containerProperties.getBatch().setMaxSize(100);
        containerProperties.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));
        return new EventHubsMessageListenerContainer(processorFactory, containerProperties);
    }
}

Event Hubs ileti başlıkları

Aşağıdaki tabloda Event Hubs ileti özelliklerinin Spring ileti üst bilgileriyle nasıl eşlendiği gösterilmektedir. Azure Event Hubs için ileti eventolarak adlandırılır.

Kayıt Dinleyicisi Modunda Event Hubs İletisi / Olay Özellikleri ile Spring İleti Üst Bilgileri Arasında Eşleme:

Event Hubs Olay Özellikleri Spring Mesaj Üstbilgi Sabitleri Tip Açıklama
Kuyruğa alınma zamanı EventHubsHeaders#ENQUEUED_TIME Anında Olayın Event Hub bölümünde kuyruğa alındığı anın UTC cinsinden değeri.
Kaydırma EventHubsHeaders#OFFSET Uzun İlişkili Olay Hub'ı bölümünden alınan olayın uzaklığı.
Bölüm anahtarı AzureHeaders#PARTITION_KEY Dize Olay ilk yayımlandığında ayarlandıysa bölüm karma anahtarı.
Bölüm Kimliği AzureHeaders#RAW_PARTITION_ID Dize Event Hub’ın bölüm kimliği.
Sıra numarası EventHubsHeaders#SEQUENCE_NUMBER Uzun Olay ilişkili Event Hub bölümünde kuyruğa alındığında ona atanan sıra numarası.
Son sıralanan olay özellikleri EventHubsHeaders#LAST_ENQUEUED_EVENT_PROPERTIES LastEnqueuedEventProperties Bu bölümde son olarak sıraya alınan olayın özellikleri.
NA AzureHeaders#CHECKPOINTER Kontrol Noktası Oluşturucu Belirli iletinin kontrol noktası üst bilgisi.

Kullanıcılar, her olayın ilgili bilgileri için ileti üst bilgilerini ayrıştırabilir. Olay için bir ileti üst bilgisi ayarlamak için, tüm özelleştirilmiş üst bilgiler bir olayın uygulama özelliği olarak konur ve burada üst bilgi özellik anahtarı olarak ayarlanır. Event Hubs'dan olaylar alındığında, tüm uygulama özellikleri ileti üst bilgisine dönüştürülür.

Not

Bölüm anahtarı, sıralanan süre, uzaklık ve sıra numarasının ileti üst bilgilerinin el ile ayarlanması desteklenmez.

Toplu tüketici modu etkinleştirildiğinde, toplu iletilerin belirli üst bilgileri aşağıda listelenmiştir; bunlar, her bir Event Hubs olayındaki değerlerin bir listesini içerir.

Toplu dinleyici modunda Event Hubs iletileri / olay özellikleri ile Spring ileti üst bilgileri arasındaki eşleme:

Event Hubs Olay Özellikleri Spring Batch Mesaj Üstbilgisi Sabitleri Tip Açıklama
Kuyruğa alınma zamanı EventHubsHeaders#ENQUEUED_TIME Anlık Liste Olay Hub'ı bölümünde her olayın ne zaman sıralandığının UTC olarak anlık listesi.
Kaydırma EventHubsHeaders#OFFSET Uzunların Listesi İlişkili Olay Hub'ı bölümünden alınan her olayın uzaklığının listesi.
Bölüm anahtarı AzureHeaders#PARTITION_KEY Dize listesi Her olayı özgün olarak yayımlarken ayarlandıysa bölüm karma anahtarının listesi.
Sıra numarası EventHubsHeaders#SEQUENCE_NUMBER Uzunların Listesi İlgili Event Hub bölümünde kuyruğa alındığında her olaya atanan sıra numaralarının listesi.
Sistem özellikleri EventHubsHeaders#BATCH_CONVERTED_SYSTEM_PROPERTIES Harita Listesi Her olayın sistem özelliklerinin listesi.
Uygulama özellikleri EventHubsHeaders#BATCH_CONVERTED_APPLICATION_PROPERTIES Harita Listesi Tüm özelleştirilmiş ileti üst bilgilerinin veya olay özelliklerinin yerleştirildiği her olayın uygulama özelliklerinin listesi.

Not

İletiler yayımlandığında, varsa yukarıdaki toplu işlem üst bilgilerinin tümü iletilerden kaldırılır.

Örnekler

Daha fazla bilgi için GitHub'daki azure-spring-boot-samples depoya bakın.

Azure Service Bus ile Spring Tümleştirmesi

Temel kavramlar

Spring Integration, Spring tabanlı uygulamalarda basit mesajlaşmaya olanak tanır ve bildirim temelli bağdaştırıcılar aracılığıyla dış sistemlerle tümleştirmeyi destekler.

Azure Service Bus için Spring Integration uzantısı projesi, Azure Service Bus için gelen ve giden kanal bağdaştırıcıları sağlar.

Not

CompletableFuture destek API'leri 2.10.0 sürümünden kullanım dışı bırakıldı ve 4.0.0 sürümünden Reactor Core ile değiştirildi. Ayrıntılar için bkz. Javadoc.

Bağımlılık kurulumu

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-servicebus</artifactId>
</dependency>

Konfigürasyon

Bu başlatıcı, yapılandırma seçeneklerinin aşağıdaki 2 bölümünü sağlar:

Bağlantı yapılandırma özellikleri

Bu bölüm, Azure Service Bus'a bağlanmak için kullanılan yapılandırma seçeneklerini içerir.

Not

Azure kaynağına erişmek için Microsoft Entra ID ile kimlik doğrulaması yapmak ve yetkilendirme sağlamak üzere bir güvenlik sorumlusu kullanmayı seçerseniz, güvenlik sorumlusuna Azure kaynağına erişmesi için yeterli izinlerin verildiğinden emin olmak üzere Microsoft Entra ID ile erişimi yetkilendirme bölümüne bakın.

spring-cloud-azure-starter-integration-servicebus bağlantısının yapılandırılabilir özellikleri:

Mülk Tip Açıklama
spring.cloud.azure.servicebus.enabled Boolean Azure Service Bus'ın etkinleştirilip etkinleştirilmediği.
spring.cloud.azure.servicebus.connection-string Dize Service Bus ad alanı bağlantı dizesi değeri.
spring.cloud.azure.servicebus.custom-endpoint-address Dize Service Bus'a bağlanırken kullanılacak özel uç nokta adresi.
spring.cloud.azure.servicebus.namespace Dize FQDN’nin ön ekini oluşturan Service Bus ad alanı değeri. FQDN, NamespaceName.DomainName dosyasından oluşmalıdır
spring.cloud.azure.servicebus.domain-name Dize Azure Service Bus ad alanı değerinin etki alanı adı.

Service Bus işlemci yapılandırma özellikleri

ServiceBusInboundChannelAdapter, iletileri tüketmek için ServiceBusProcessorClient kullanır; bir ServiceBusProcessorClient öğesinin genel özelliklerini yapılandırmak için geliştiriciler yapılandırmada ServiceBusContainerProperties kullanabilir. ile nasıl çalışılacağına dair aşağıdaki bölüme bakın.

Temel kullanım

Azure Service Bus'a ileti gönderme

  1. Kimlik bilgisi yapılandırma seçeneklerini doldurun.

    • Bağlantı dizesi olarak kimlik bilgileri için application.yml dosyanızda aşağıdaki özellikleri yapılandırın:

      spring:
  cloud:
    azure:
      servicebus:
        connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}

      Not

      Microsoft, kullanılabilir en güvenli kimlik doğrulama akışının kullanılmasını önerir. Veritabanları, önbellekler, mesajlaşma veya yapay zeka hizmetleri gibi bu yordamda açıklanan kimlik doğrulama akışı, uygulamaya çok yüksek düzeyde güven gerektirir ve diğer akışlarda mevcut olmayan riskler taşır. Bu akışı yalnızca parolasız veya anahtarsız bağlantılar için yönetilen kimlikler gibi daha güvenli seçenekler uygun olmadığında kullanın. Yerel makine işlemleri için parolasız veya anahtarsız bağlantılar için kullanıcı kimliklerini tercih edin.

    • Yönetilen kimlik olarak kimlik bilgileri için application.yml dosyanızda aşağıdaki özellikleri yapılandırın:

      spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_CLIENT_ID}
      profile:
        tenant-id: <tenant>
      servicebus:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

Not

tenant-id için izin verilen değerler şunlardır: common, organizations, consumersveya kiracı kimliği. Bu değerler hakkında daha fazla bilgi için, Hata AADSTS50020 - Kimlik sağlayıcısındaki kullanıcı hesabı kiracısında mevcut değil başlıklı makalenin Yanlış uç nokta kullanıldı (kişisel ve kurumsal hesaplar) bölümüne bakın. Tek kiracılı uygulamanızı dönüştürme hakkında bilgi için bkz. Tek kiracılı uygulamayı Microsoft Entra IDüzerinde çok kiracılıya dönüştürme.

  • Hizmet sorumlusu olarak kimlik bilgileri için application.yml dosyanızda aşağıdaki özellikleri yapılandırın:

    spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      servicebus:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

Not

tenant-id için izin verilen değerler şunlardır: common, organizations, consumersveya kiracı kimliği. Bu değerler hakkında daha fazla bilgi için, Hata AADSTS50020 - Kimlik sağlayıcısındaki kullanıcı hesabı kiracıda mevcut değil başlıklı konunun Yanlış uç nokta kullanıldı (kişisel ve kuruluş hesapları) bölümüne bakın. Tek kiracılı uygulamanızı dönüştürme hakkında bilgi için bkz. Tek kiracılı uygulamayı Microsoft Entra IDüzerinde çok kiracılıya dönüştürme.

  1. Service Bus'a ileti göndermek için DefaultMessageHandler bean ile ServiceBusTemplate oluşturun, ServiceBusTemplate için varlık türünü ayarlayın. Bu örnekte örnek olarak Service Bus Kuyruğu alınıyor.

    class Demo {
    private static final String OUTPUT_CHANNEL = "queue.output";

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public MessageHandler queueMessageSender(ServiceBusTemplate serviceBusTemplate) {
        serviceBusTemplate.setDefaultEntityType(ServiceBusEntityType.QUEUE);
        DefaultMessageHandler handler = new DefaultMessageHandler(QUEUE_NAME, serviceBusTemplate);
        handler.setSendCallback(new ListenableFutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                LOGGER.info("Message was sent successfully.");
            }

            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error("There was an error sending the message.", ex);
            }
        });

        return handler;
    }
}

  2. Bir ileti kanalı aracılığıyla yukarıdaki ileti işleyicisiyle bir ileti ağ geçidi bağlaması oluşturun.

    class Demo {
    @Autowired
    QueueOutboundGateway messagingGateway;

    @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
    public interface QueueOutboundGateway {
        void send(String text);
    }
}

  3. Ağ geçidini kullanarak ileti gönderme.

    class Demo {
    public void demo() {
        this.messagingGateway.send(message);
    }
}

Azure Service Bus'tan ileti alma

  1. Kimlik bilgisi yapılandırma seçeneklerini doldurun.

  2. Giriş kanalı olarak bir ileti kanalı çekirdeği oluşturun.

    @Configuration
class Demo {
    private static final String INPUT_CHANNEL = "input";

    @Bean
    public MessageChannel input() {
        return new DirectChannel();
    }
}

  3. Service Bus’tan iletiler almak için ServiceBusMessageListenerContainer bean’i ile ServiceBusInboundChannelAdapter oluşturun. Bu örnekte örnek olarak Service Bus Kuyruğu alınıyor.

    @Configuration
class Demo {
    private static final String QUEUE_NAME = "queue1";

    @Bean
    public ServiceBusMessageListenerContainer messageListenerContainer(ServiceBusProcessorFactory processorFactory) {
        ServiceBusContainerProperties containerProperties = new ServiceBusContainerProperties();
        containerProperties.setEntityName(QUEUE_NAME);
        containerProperties.setAutoComplete(false);
        return new ServiceBusMessageListenerContainer(processorFactory, containerProperties);
    }

    @Bean
    public ServiceBusInboundChannelAdapter queueMessageChannelAdapter(
        @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
        ServiceBusMessageListenerContainer listenerContainer) {
        ServiceBusInboundChannelAdapter adapter = new ServiceBusInboundChannelAdapter(listenerContainer);
        adapter.setOutputChannel(inputChannel);
        return adapter;
    }
}

  4. Daha önce oluşturduğumuz ileti kanalı aracılığıyla ServiceBusInboundChannelAdapter ile bir ileti alıcı bağlaması oluşturun.

    class Demo {
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
        String message = new String(payload);
        LOGGER.info("New message received: '{}'", message);
        checkpointer.success()
                .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
                .doOnError(e -> LOGGER.error("Error found", e))
                .block();
    }
}

ObjectMapper'ı özelleştirmek için ServiceBusMessageConverter'ı yapılandırma

ServiceBusMessageConverter, kullanıcıların ObjectMapper özelleştirmesine olanak tanımak için yapılandırılabilir bir bean olarak tasarlanmıştır.

Service Bus ileti başlıkları

Birden çok Spring üst bilgisi sabitine eşlenebilen bazı Service Bus üst bilgileri için, farklı Spring üst bilgilerinin önceliği listelenir.

Service Bus Başlıkları ile Spring Başlıkları Arasındaki Eşleme:

Service Bus mesaj üstbilgileri ve özellikleri Spring ileti üst bilgisi sabitleri Tip Yapılandırılabilir Açıklama
İçerik türü MessageHeaders#CONTENT_TYPE Dize Evet İletinin RFC2045 İçerik Türü tanımlayıcısı.
Korelasyon Kimliği ServiceBusMessageHeaders#CORRELATION_ID Dize Evet İletinin korelasyon kimliği
İleti Kimliği ServiceBusMessageHeaders#MESSAGE_ID Dize Evet İletinin mesaj kimliği, bu üst bilgi MessageHeaders#ID öğesinden daha yüksek önceliğe sahiptir.
İleti Kimliği MessageHeaders#ID UUID Evet İletinin ileti kimliği, bu üst bilgi ServiceBusMessageHeaders#MESSAGE_ID'dan daha düşük önceliğe sahiptir.
Bölüm anahtarı ServiceBusMessageHeaders#PARTITION_KEY Dize Evet İletiyi bölümlenmiş bir varlığa göndermek için bölüm anahtarı.
Yanıtla MessageHeaders#REPLY_CHANNEL Dize Evet Yanıtların gönderleneceği varlığın adresi.
Oturum kimliğine yanıt ver ServiceBusMessageHeaders#REPLY_TO_SESSION_ID Dize Evet İletinin ReplyToGroupId özelliğinin değeri.
Zamanlanmış kuyruğa alma zamanı UTC ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME OffsetDateTime Evet İletinin Service Bus'ta kuyruğa alınması gereken tarih ve saat; bu üst bilgi, AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE öğesinden daha yüksek önceliğe sahiptir.
Zamanlanmış kuyruğa alma zamanı (UTC) AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE Tam sayı Evet Service Bus'ta iletinin kuyruğa alınması gerektiği tarih ve saat; bu üstbilgi, ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME öğesinden daha düşük önceliğe sahiptir.
Oturum Kimliği ServiceBusMessageHeaders#SESSION_ID Dize Evet Oturum farkında bir varlık için oturum kimliği.
Yaşam süresi ServiceBusMessageHeaders#TIME_TO_LIVE Süre Evet Bu iletinin süresi dolana kadar kalan süre.
Alıcı ServiceBusMessageHeaders#TO Dize Evet İletinin "to" adresi; yönlendirme senaryolarında gelecekte kullanılmak üzere ayrılmıştır ve şu anda aracının kendisi tarafından yok sayılmaktadır.
Konu ServiceBusMessageHeaders#SUBJECT Dize Evet İletinin konusu.
Teslim edilemeyen ileti hata açıklaması ServiceBusMessageHeaders#DEAD_LETTER_ERROR_DESCRIPTION Dize Hayır Teslim edilemeyen iletinin açıklaması.
Teslim edilemeyen mektup nedeni ServiceBusMessageHeaders#DEAD_LETTER_REASON Dize Hayır Bir iletinin yazılmama nedeni.
Ölü ileti kaynağı ServiceBusMessageHeaders#DEAD_LETTER_SOURCE Dize Hayır İletinin teslim edilmediği varlık.
Teslim sayısı ServiceBusMessageHeaders#DELIVERY_COUNT uzun Hayır Bu iletinin istemcilere teslim edildiği sayısı.
Sıralanmış sıra numarası ServiceBusMessageHeaders#ENQUEUED_SEQUENCE_NUMBER uzun Hayır Service Bus tarafından bir iletiye atanan sıralı sıra numarası.
Kuyruğa alınma zamanı ServiceBusMessageHeaders#ENQUEUED_TIME OffsetDateTime Hayır Bu iletinin Service Bus'ta kuyruğa alındığı tarih ve saat.
Son kullanma tarihi: ServiceBusMessageHeaders#EXPIRES_AT OffsetDateTime Hayır Bu iletinin süresinin dolacağı tarih ve saat.
Kilit belirteci ServiceBusMessageHeaders#LOCK_TOKEN Dize Hayır Geçerli ileti için kilit belirteci.
Şu zamana kadar kilitli: ServiceBusMessageHeaders#LOCKED_UNTIL OffsetDateTime Hayır Bu iletinin kilidinin sona ereceği tarih ve saat.
Sıra numarası ServiceBusMessageHeaders#SEQUENCE_NUMBER uzun Hayır Service Bus tarafından bir iletiye atanan benzersiz numara.
Devlet ServiceBusMessageHeaders#STATE ServiceBusMessageState Hayır Aktif, Ertelenmiş veya Zamanlanmış durumunda olabilen iletinin durumu.

Bölüm anahtarı desteği

Bu başlatıcı, ileti üst bilgisinde bölüm anahtarı ve oturum kimliği ayarlamaya olanak tanıyarak Service Bus bölümlemeyi destekler. Bu bölümde, iletiler için bölüm anahtarının nasıl ayarlanacağı anlatlenmektedir.

Önerilen: ServiceBusMessageHeaders.PARTITION_KEY öğesini üst bilginin anahtarı olarak kullanın.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(ServiceBusMessageHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Önerilmez, ancak şu anda desteklenmektedir: üst bilgi anahtarı olarak AzureHeaders.PARTITION_KEY.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(AzureHeaders.PARTITION_KEY, "Customize partition key")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Not

İleti üst bilgilerinde hem ServiceBusMessageHeaders.PARTITION_KEY hem de AzureHeaders.PARTITION_KEY ayarlandığında, ServiceBusMessageHeaders.PARTITION_KEY tercih edilir.

Oturum desteği

Bu örnekte, uygulamadaki bir iletinin oturum kimliğinin el ile nasıl ayarlanacağı gösterilmektedir.

public class SampleController {
    @PostMapping("/messages")
    public ResponseEntity<String> sendMessage(@RequestParam String message) {
        LOGGER.info("Going to add message {} to Sinks.Many.", message);
        many.emitNext(MessageBuilder.withPayload(message)
                                    .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session ID")
                                    .build(), Sinks.EmitFailureHandler.FAIL_FAST);
        return ResponseEntity.ok("Sent!");
    }
}

Not

İleti üst bilgilerinde ServiceBusMessageHeaders.SESSION_ID ayarlanmışsa ve farklı bir ServiceBusMessageHeaders.PARTITION_KEY üst bilgisi de ayarlanmışsa, nihayetinde oturum kimliğinin değeri bölüm anahtarının değerinin üzerine yazmak için kullanılır.

Service Bus istemci özelliklerini özelleştirme

Geliştiriciler Service Bus İstemcisi özelliklerini özelleştirmek için AzureServiceClientBuilderCustomizer kullanabilir. Aşağıdaki örnek, sessionIdleTimeoutiçindeki ServiceBusClientBuilder özelliğini özelleştirmektedir:

@Bean
public AzureServiceClientBuilderCustomizer<ServiceBusClientBuilder.ServiceBusSessionProcessorClientBuilder> customizeBuilder() {
    return builder -> builder.sessionIdleTimeout(Duration.ofSeconds(10));
}

Örnekler

Daha fazla bilgi için GitHub'daki azure-spring-boot-samples depoya bakın.

Azure Depolama Kuyruğu ile Spring Tümleştirmesi

Temel kavramlar

Azure Kuyruk Depolama, çok sayıda iletiyi depolamaya yönelik bir hizmettir. HTTP veya HTTPS kullanarak kimliği doğrulanmış çağrılar aracılığıyla dünyanın herhangi bir yerinden iletilere erişebilirsiniz. Kuyruk iletisinin boyutu en fazla 64 KB olabilir. Kuyruk, depolama hesabının toplam kapasite sınırına kadar milyonlarca ileti içerebilir. Kuyruklar, işlerin eşzamansız olarak işlenmesi için bir iş birikimi oluşturmak amacıyla yaygın olarak kullanılır.

Bağımlılık kurulumu

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter-integration-storage-queue</artifactId>
</dependency>

Konfigürasyon

Bu başlatıcı aşağıdaki yapılandırma seçeneklerini sağlar:

Bağlantı yapılandırma özellikleri

Bu bölüm, Azure Depolama Kuyruğu'na bağlanmak için kullanılan yapılandırma seçeneklerini içerir.

Not

Azure kaynağına erişmek için Microsoft Entra ID ile kimlik doğrulaması yapmak ve yetkilendirme sağlamak üzere bir güvenlik sorumlusu kullanmayı seçerseniz, güvenlik sorumlusuna Azure kaynağına erişmesi için yeterli izinlerin verildiğinden emin olmak üzere Microsoft Entra ID ile erişimi yetkilendirme bölümüne bakın.

spring-cloud-azure-starter-integration-storage-queue bağlantısının yapılandırılabilir özellikleri:

Mülk Tip Açıklama
spring.cloud.azure.storage.queue.enabled Boolean Azure Depolama Kuyruğu'nın etkinleştirilip etkinleştirilmediği.
spring.cloud.azure.storage.queue.connection-string Dize Depolama Kuyruğu ad alanı bağlantı dizesi değeri.
spring.cloud.azure.storage.queue.accountName Dize Depolama Kuyruğu hesap adı.
spring.cloud.azure.storage.queue.accountKey Dize Depolama Kuyruğu hesap anahtarı.
spring.cloud.azure.storage.queue.endpoint Dize Depolama Kuyruğu hizmet uç noktası.
spring.cloud.azure.storage.queue.sasToken Dize SAS belirteç kimlik bilgisi
spring.cloud.azure.storage.queue.serviceVersion QueueServiceVersion API istekleri yapılırken kullanılan QueueServiceVersion değeri.
spring.cloud.azure.storage.queue.messageEncoding Dize Kuyruk iletisi kodlama.

Temel kullanım

Azure Depolama Kuyruğuna ileti gönderme

  1. Kimlik bilgisi yapılandırma seçeneklerini doldurun.

    • Bağlantı dizesi olarak kimlik bilgileri için application.yml dosyanızda aşağıdaki özellikleri yapılandırın:

      spring:
  cloud:
    azure:
      storage:
        queue:
          connection-string: ${AZURE_STORAGE_QUEUE_CONNECTION_STRING}

      Not

      Microsoft, kullanılabilir en güvenli kimlik doğrulama akışının kullanılmasını önerir. Veritabanları, önbellekler, mesajlaşma veya yapay zeka hizmetleri gibi bu yordamda açıklanan kimlik doğrulama akışı, uygulamaya çok yüksek düzeyde güven gerektirir ve diğer akışlarda mevcut olmayan riskler taşır. Bu akışı yalnızca parolasız veya anahtarsız bağlantılar için yönetilen kimlikler gibi daha güvenli seçenekler uygun olmadığında kullanın. Yerel makine işlemleri için parolasız veya anahtarsız bağlantılar için kullanıcı kimliklerini tercih edin.

    • Yönetilen kimlik olarak kimlik bilgileri için application.yml dosyanızda aşağıdaki özellikleri yapılandırın:

      spring:
  cloud:
    azure:
      credential:
        managed-identity-enabled: true
        client-id: ${AZURE_CLIENT_ID}
      profile:
        tenant-id: <tenant>
      storage:
        queue:
          account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}

Not

tenant-id için izin verilen değerler şunlardır: common, organizations, consumersveya kiracı kimliği. Bu değerler hakkında daha fazla bilgi için, Hata AADSTS50020 - Kimlik sağlayıcısındaki kullanıcı hesabı kiracıda mevcut değil başlıklı konunun Yanlış uç nokta kullanıldı (kişisel ve kuruluş hesapları) bölümüne bakın. Tek kiracılı uygulamanızı dönüştürme hakkında bilgi için bkz. Tek kiracılı uygulamayı Microsoft Entra IDüzerinde çok kiracılıya dönüştürme.

  • Hizmet sorumlusu olarak kimlik bilgileri için application.yml dosyanızda aşağıdaki özellikleri yapılandırın:

    spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: <tenant>
      storage:
        queue:
          account-name: ${AZURE_STORAGE_QUEUE_ACCOUNT_NAME}

Not

tenant-id için izin verilen değerler şunlardır: common, organizations, consumersveya kiracı kimliği. Bu değerler hakkında daha fazla bilgi için, Hata AADSTS50020 - Kimlik sağlayıcısındaki kullanıcı hesabı kiracıda mevcut değil başlıklı konunun Yanlış uç nokta kullanıldı (kişisel ve kuruluş hesapları) bölümüne bakın. Tek kiracılı uygulamanızı dönüştürme hakkında bilgi için bkz. Tek kiracılı uygulamayı Microsoft Entra IDüzerinde çok kiracılıya dönüştürme.

  1. Depolama Kuyruğuna ileti göndermek için DefaultMessageHandler çekirdeğiyle StorageQueueTemplate oluşturun.

    class Demo {
    private static final String STORAGE_QUEUE_NAME = "example";
    private static final String OUTPUT_CHANNEL = "output";

    @Bean
    @ServiceActivator(inputChannel = OUTPUT_CHANNEL)
    public MessageHandler messageSender(StorageQueueTemplate storageQueueTemplate) {
        DefaultMessageHandler handler = new DefaultMessageHandler(STORAGE_QUEUE_NAME, storageQueueTemplate);
        handler.setSendCallback(new ListenableFutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                LOGGER.info("Message was sent successfully.");
            }

            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error("There was an error sending the message.", ex);
            }
        });
        return handler;
    }
}

  2. Bir ileti kanalı aracılığıyla yukarıdaki ileti işleyicisiyle bir İleti ağ geçidi bağlaması oluşturun.

    class Demo {
    @Autowired
    StorageQueueOutboundGateway storageQueueOutboundGateway;

    @MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
    public interface StorageQueueOutboundGateway {
        void send(String text);
    }
}

  3. Ağ geçidini kullanarak ileti gönderme.

    class Demo {
    public void demo() {
        this.storageQueueOutboundGateway.send(message);
    }
}

Azure Depolama Kuyruğu'ndan ileti alma

  1. Kimlik bilgisi yapılandırma seçeneklerini doldurun.

  2. Giriş kanalı olarak bir ileti kanalı çekirdeği oluşturun.

    class Demo {
    private static final String INPUT_CHANNEL = "input";

    @Bean
    public MessageChannel input() {
        return new DirectChannel();
    }
}

  3. Depolama Kuyruğu'na mesaj almak için StorageQueueTemplate bean'i ile StorageQueueMessageSource oluşturun.

    class Demo {
    private static final String STORAGE_QUEUE_NAME = "example";

    @Bean
    @InboundChannelAdapter(channel = INPUT_CHANNEL, poller = @Poller(fixedDelay = "1000"))
    public StorageQueueMessageSource storageQueueMessageSource(StorageQueueTemplate storageQueueTemplate) {
        return new StorageQueueMessageSource(STORAGE_QUEUE_NAME, storageQueueTemplate);
    }
}

  4. Daha önce oluşturduğumuz ileti kanalı aracılığıyla son adımda oluşturulan StorageQueueMessageSource ile bir ileti alıcısı bağlaması oluşturun.

    class Demo {
    @ServiceActivator(inputChannel = INPUT_CHANNEL)
    public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
        String message = new String(payload);
        LOGGER.info("New message received: '{}'", message);
        checkpointer.success()
            .doOnError(Throwable::printStackTrace)
            .doOnSuccess(t -> LOGGER.info("Message '{}' successfully checkpointed", message))
            .block();
    }
}

Örnekler

Daha fazla bilgi için GitHub'daki azure-spring-boot-samples depoya bakın.

  • Last updated on