Spring Cloud Azure-támogatás a Spring Integrationhez

Az Azure Spring Integration Extension spring integration adaptereket biztosít a Azure SDK for Javakülönböző szolgáltatásaihoz. Spring Integration-támogatást biztosítunk ezekhez az Azure-szolgáltatásokhoz: Event Hubs, Service Bus, Storage Queue. A támogatott adapterek listája a következő:

• spring-cloud-azure-starter-integration-eventhubs – további információ: Spring Integration with Azure Event Hubs
• spring-cloud-azure-starter-integration-servicebus – további információ: Spring Integration with Azure Service Bus
• spring-cloud-azure-starter-integration-storage-queue – további információ: Spring Integration with Azure Storage Queue

Tavaszi integráció az Azure Event Hubsszal

Főbb fogalmak

Az Azure Event Hubs egy big data streamelési platform és eseménybetöltési szolgáltatás. Másodpercenként több millió esemény fogadására és feldolgozására képes. Az eseményközpontba küldött adatok átalakíthatók és tárolhatók bármely valós idejű elemzési szolgáltató vagy kötegelési/tárolási adapter használatával.

A Spring Integration lehetővé teszi az egyszerű üzenetkezelést a Spring-alapú alkalmazásokban, és deklaratív adaptereken keresztül támogatja a külső rendszerekkel való integrációt. Ezek az adapterek magasabb szintű absztrakciós réteget biztosítanak a Spring távoli kommunikációhoz, üzenetkezeléshez és ütemezéshez nyújtott támogatása fölött. Az Spring Integration for Event Hubs bővítményprojekt bejövő és kimenő csatornaadaptereket és átjárókat biztosít az Azure Event Hubshoz.

Jegyzet

Az RxJava támogatási API-k a 4.0.0-s verzióról törlődnek. Részletekért lásd: Javadoc.

Fogyasztói csoport

Az Event Hubs hasonló támogatást nyújt a fogyasztói csoportoknak, mint az Apache Kafka, de kismértékben eltérő logikával. Bár a Kafka az összes commitált offsetet a brokerben tárolja, az éppen feldolgozás alatt álló Event Hubs-üzenetek offsetjeit manuálisan kell tárolni. Az Event Hubs SDK az ilyen eltolások Azure Storage-ban való tárolására szolgál.

Particionálás támogatása

Az Event Hubs a Kafkához hasonló fizikai partíciót biztosít. A Kafka felhasználói és partíciói közötti automatikus újraelosztással ellentétben azonban az Event Hubs egyfajta megelőző módot biztosít. A tárfiók bérletként működik annak meghatározásához, hogy melyik partíció melyik felhasználó tulajdonában van. Amikor egy új felhasználó elindul, megpróbál ellopni néhány partíciót a legtöbb nehéz terhelésű felhasználótól a számítási feladatok kiegyensúlyozása érdekében.

A terheléselosztási stratégia megadásához a fejlesztők EventHubsContainerProperties használhatnak a konfigurációhoz. A konfigurálására vonatkozó példát lásd a következő szakaszban.

Kötegelt ügyféltámogatás

A EventHubsInboundChannelAdapter támogatja a kötegfogyasztó módot. Az engedélyezéshez a felhasználók ListenerMode.BATCH figyelő módot adhatnak meg EventHubsInboundChannelAdapter-példányok létrehozásakor. Ha engedélyezve van, egy üzenet, amelynek hasznos adata a kötegelt események listája, a rendszer megkapja és átadja az alsóbb rétegbeli csatornának. Minden üzenetfejléc szintén listává alakul, amelynek tartalma az egyes eseményekből kinyert, hozzá tartozó fejlécérték. A partícióazonosító, az ellenőrzőpont és az utolsó lekérdezett tulajdonságok közösségi fejlécei egyetlen értékként jelennek meg a teljes eseménykötegben, amelyek ugyanazt a tulajdonságot használják. További információ: Event Hubs-üzenetfejlécek szakasz.

Jegyzet

Az ellenőrzőpont fejléce csak akkor létezik, ha MANUÁLIS ellenőrzőpont mód van használatban.

A kötegfelhasználók ellenőrzőpontozása két módot támogat: BATCH és MANUAL. BATCH mód egy automatikus ellenőrzőpontozási mód, amely az események teljes kötegét együttesen ellenőrzi a beérkezés után. MANUAL mód célja a felhasználók által generált események ellenőrzőpontként való mentése. Használat esetén a Ellenőrzőpont az üzenet fejlécébe kerül, és a felhasználók használhatják az ellenőrzőpont-ellenőrzést.

A kötegfogyasztási házirend a max-size és max-wait-time tulajdonságaival adható meg, ahol a max-size kötelező, a max-wait-time pedig opcionális tulajdonság. A kötegfeldolgozási stratégia megadásához a fejlesztők a konfigurációban a EventHubsContainerProperties elemet használhatják. Lásd a következő szakaszt, ahol példát talál arra, hogyan konfigurálhatja a elemet.

Függőségek beállítása

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

Konfiguráció

Ez az alapindító a következő 3 konfigurációs lehetőséget biztosítja:

Kapcsolatkonfiguráció tulajdonságai

Ez a szakasz az Azure Event Hubshoz való csatlakozáshoz használt konfigurációs beállításokat tartalmazza.

Jegyzet

Ha biztonsági tagot használ egy Azure-erőforrás eléréséhez a Microsoft Entra-azonosítóval történő hitelesítéshez és engedélyezéshez, tekintse meg Hozzáférés engedélyezése a Microsoft Entra-azonosítóval annak biztosításához, hogy a biztonsági tag megkapta-e a megfelelő engedélyt az Azure-erőforrás eléréséhez.

A kapcsolat konfigurálható tulajdonságai spring-cloud-azure-starter-integration-eventhubs:

Ingatlan	Típus	Leírás
spring.cloud.azure.eventhubs.enabled	logikai érték	Az Azure Event Hubs engedélyezése.
spring.cloud.azure.eventhubs.connection-string	karakterlánc	Az Event Hubs-névtér kapcsolati karakterláncának értéke.
spring.cloud.azure.eventhubs.namespace	karakterlánc	Az Event Hubs névtér értéke, amely az FQDN előtagja. Az FQDN-nek a NamespaceName.DomainName elemekből kell állnia
spring.cloud.azure.eventhubs.domain-name	karakterlánc	Egy Azure Event Hubs-névtérérték tartományneve.
spring.cloud.azure.eventhubs.custom-endpoint-address	karakterlánc	Egyéni végpont címe.
spring.cloud.azure.eventhubs.shared-connection	logikai	Azt, hogy a mögöttes EventProcessorClient és az EventHubProducerAsyncClient ugyanazt a kapcsolatot használja-e. Alapértelmezés szerint minden létrehozott Event Hub-ügyfélhez új kapcsolat jön létre, és azt használja a rendszer.

Ellenőrzőpont konfigurációs tulajdonságai

Ez a szakasz a Storage Blobs szolgáltatás konfigurációs beállításait tartalmazza, amely a partíció tulajdonjogának és ellenőrzőpont-adatainak megőrzésére szolgál.

Jegyzet

A 4.0.0-s verzióból, ha a tulajdonság spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists nincs manuálisan engedélyezve, a rendszer nem hoz létre automatikusan tárolótárolót.

A(z) spring-cloud-azure-starter-integration-eventhubs konfigurálható tulajdonságainak ellenőrzőpontba mentése:

Ingatlan	Típus	Leírás
spring.cloud.azure.eventhubs.processor.checkpoint-store.create-container-if-not-exists	logikai	Engedélyezi-e a tárolók létrehozását, ha nem létezik.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-name	karakterlánc	A tárfiók neve.
spring.cloud.azure.eventhubs.processor.checkpoint-store.account-key	karakterlánc	Tárfiók hozzáférési kulcsa.
spring.cloud.azure.eventhubs.processor.checkpoint-store.container-name	karakterlánc	Tárolókonténer neve.

Az Azure Service SDK gyakori konfigurációs beállításai a Storage Blob CheckPoint Store-hoz is konfigurálhatók. A támogatott konfigurációs beállításokat a Spring Cloud Azure konfiguráció mutatja be, és az egységes spring.cloud.azure. előtaggal vagy a spring.cloud.azure.eventhubs.processor.checkpoint-store. előtaggal konfigurálhatók.

Az Event Hub processzorkonfigurációs tulajdonságai

A EventHubsInboundChannelAdapter a EventProcessorClient használja egy eseményközpont üzeneteinek felhasználására, egy EventProcessorClientáltalános tulajdonságainak konfigurálására, a fejlesztők EventHubsContainerProperties használhatnak a konfigurációhoz. A használatáról lásd a következő szakaszt.

Alapszintű használat

Üzenetek küldése az Azure Event Hubsba

1. Adja meg a hitelesítő adatok konfigurációs beállításait.

   • A hitelesítő adatok kapcsolati sztringként való megadásához konfigurálja a következő tulajdonságokat a application.yml fájlban:

     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}
     

     Jegyzet

     A Microsoft az elérhető legbiztonságosabb hitelesítési folyamat használatát javasolja. Az ebben az eljárásban ismertetett hitelesítési folyamat, például adatbázisok, gyorsítótárak, üzenetkezelés vagy AI-szolgáltatások esetében, nagyon nagy megbízhatóságot igényel az alkalmazásban, és más folyamatokban nem jelenik meg kockázattal. Ezt a folyamatot csak akkor használja, ha a biztonságosabb lehetőségek, például a jelszó nélküli vagy kulcs nélküli kapcsolatokhoz használható felügyelt identitások, nem megvalósíthatók. A helyi gépi műveletekhez előnyben részesítse a jelszó nélküli vagy kulcs nélküli kapcsolatok felhasználói identitásait.

   • A hitelesítő adatok felügyelt identitásként való megadásához konfigurálja a következő tulajdonságokat a application.yml fájlban:

     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}
     

   • A szolgáltatásnév-azonosítóként használt hitelesítő adatokhoz állítsa be a következő tulajdonságokat a application.yml fájlban:

     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}
     

Jegyzet

A tenant-id engedélyezett értékek a következők: common, organizations, consumersvagy bérlőazonosító. Ezekről az értékekről további információt az AADSTS50020 hiba – Az identitásszolgáltatótól származó felhasználói fiók nem létezik a bérlőben című cikk Rossz végpont (személyes és szervezeti fiókok) szakaszában talál. Az egybérlős alkalmazás átalakításával kapcsolatos információkért lásd: Egybérlős alkalmazás átalakítása több-bérlős alkalmazássá a Microsoft Entra ID-ban.

1. A(z) EventHubsTemplate bean használatával hozzon létre egy DefaultMessageHandler elemet, hogy üzeneteket küldhessen az Event Hubsba.

   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. Hozzon létre egy üzenetátjáró-kötést a fenti üzenetkezelővel egy üzenetcsatornán keresztül.

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

3. Üzenetek küldése az átjárón keresztül.

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

Üzenetek fogadása az Azure Event Hubsból

1. Adja meg a hitelesítő adatok konfigurációs beállításait.

2. Definiáljon egy üzenetcsatorna-beant, amely bemeneti csatornaként szolgál.

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

3. Hozza létre a EventHubsInboundChannelAdapter elemet a EventHubsMessageListenerContainer bean használatával az Event Hubsból érkező üzenetek fogadásához.

   @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. Hozzon létre egy üzenet fogadó kötést az EventHubsInboundChannelAdapterrel a korábban létrehozott üzenetcsatornán keresztül.

   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();
       }
   }
   

Az EventHubsMessageConverter konfigurálása az objectMapper testreszabásához

EventHubsMessageConverter konfigurálható bean-ként úgy lett kialakítva, hogy a felhasználók testre szabhassák az ObjectMappert.

Kötegelt feldolgozást végző fogyasztók támogatása

Az Event Hubsból származó üzenetek köteges fogyasztása hasonló a fenti példához, emellett a EventHubsInboundChannelAdapter esetében be kell állítani a köteges fogyasztáshoz kapcsolódó konfigurációs beállításokat.

A EventHubsInboundChannelAdapterlétrehozásakor a figyelő üzemmódot BATCHkell beállítani. A EventHubsMessageListenerContainer létrehozásakor állítsa az ellenőrzőpont módját MANUAL-ra vagy BATCH-re, a kötegbeállítások pedig szükség szerint konfigurálhatók.

@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);
    }
}

Az Event Hubs üzenetfejlécei

Az alábbi táblázat bemutatja, hogyan feleltethetők meg az Event Hubs üzenettulajdonságai a Spring üzenetfejléceinek. Az Azure Event Hubs esetében az üzenet neve event.

Az Event Hubs üzenet-/eseménytulajdonságai és a Spring-üzenetfejlécek közötti megfeleltetés rekordfigyelő üzemmódban:

Az Event Hubs eseménytulajdonságai	Tavaszi üzenetfejléc-állandók	Típus	Leírás
Várólista idő	EventHubsHeaders#ENQUEUED_TIME	Azonnali	Az UTC szerinti időpont, amikor az eseményt sorba állították az Event Hub-partícióban.
Ellensúlyoz	EventHubsHeaders#OFFSET	Hosszú	Az esemény offsetje a társított Event Hub-partícióból való beérkezésekor.
partíciós kulcs	AzureHeaders#PARTITION_KEY	karakterlánc	A particionálási hashelési kulcs, ha az esemény közzétételekor be lett állítva.
Partícióazonosító	AzureHeaders#RAW_PARTITION_ID	karakterlánc	Az Event Hub partícióazonosítója.
Sorozatszáma	EventHubsHeaders#SEQUENCE_NUMBER	Hosszú	Az eseményhez rendelt sorszám, amikor az a társított Event Hub-partícióba sorba került.
Az utoljára várólistára helyezett esemény tulajdonságai	EventHubsHeaders#LAST_ENQUEUED_EVENT_PROPERTIES	Az utoljára üzenetsorba helyezett esemény tulajdonságai	A partícióban utoljára sorba állított esemény tulajdonságai.
N/A	AzureHeaders#CHECKPOINTER	Ellenőrzőpont	Az adott üzenet ellenőrzőpontjának fejléce.

A felhasználók elemezhetik az egyes események kapcsolódó információinak üzenetfejléceit. Az esemény üzenetfejlécének beállításához a program az összes testreszabott fejlécet egy esemény alkalmazástulajdonságaként helyezi el, ahol a fejléc tulajdonságkulcsként van beállítva. Amikor események érkeznek az Event Hubstól, az összes alkalmazástulajdonság az üzenetfejlécre lesz konvertálva.

Megjegyzés

A partíciókulcs, a sorba állítási idő, az offset és a sorszám üzenetfejlécei nem állíthatók be manuálisan.

Ha a batch-consumer mód engedélyezve van, a kötegelt üzenetek specifikus fejlécei az alábbiak szerint vannak felsorolva; ezek mindegyike az egyes Event Hubs-eseményekből származó értékek listáját tartalmazza.

Leképezés az Event Hubs-üzenet/eseménytulajdonságok és a Spring Message-fejlécek között Batch-figyelő módban:

Az Event Hubs eseménytulajdonságai	Spring Batch üzenetfejléc-állandók	Típus	Leírás
Várólista idő	EventHubsHeaders#ENQUEUED_TIME	Azonnali lista	Annak a pillanatnak a listája (UTC), amikor az egyes eseményeket az Event Hub partícióba foglalták.
Ellensúlyoz	EventHubsHeaders#OFFSET	Hosszú lista	Az egyes események eltolásainak listája a hozzá tartozó Event Hub-partícióból való beérkezésükkor.
partíciós kulcs	AzureHeaders#PARTITION_KEY	Sztringek listája	A partíció kivonatolási kulcsának listája, ha az az egyes események eredeti közzétételekor lett beállítva.
Sorozatszáma	EventHubsHeaders#SEQUENCE_NUMBER	A hosszúak listája	Az egyes eseményekhez rendelt sorszámok listája, amikor az eseményt a társított Event Hub-partícióban lekérdezték.
Rendszertulajdonságok	EventHubsHeaders#BATCH_CONVERTED_SYSTEM_PROPERTIES	Térképlista	Az egyes események rendszertulajdonságainak listája.
Alkalmazástulajdonságok	EventHubsHeaders#BATCH_CONVERTED_APPLICATION_PROPERTIES	Térképlista	Az egyes események alkalmazástulajdonságainak listája, ahol az összes testreszabott üzenetfejléc vagy eseménytulajdonság el lesz helyezve.

Megjegyzés

Üzenetek közzétételekor a fenti kötegfejlécek el lesznek távolítva az üzenetekből, ha léteznek.

Minták

További információt a azure-spring-boot-samples GitHub adattárában talál.

Spring integráció az Azure Service Bus-szal

Főbb fogalmak

A Spring Integration lehetővé teszi az egyszerű üzenetkezelést a Spring-alapú alkalmazásokban, és deklaratív adaptereken keresztül támogatja a külső rendszerekkel való integrációt.

Az Azure Service Bus Spring Integration bővítményprojektje bejövő és kimenő csatornaadaptereket biztosít az Azure Service Bushoz.

Jegyzet

Az CompletableFuture támogatási API-k elavultak a 2.10.0-s verzióról, és a Reactor Core a 4.0.0-s verzióról váltja fel. Részletekért lásd: Javadoc.

Függőségek beállítása

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

Konfiguráció

Ez az alapindító a következő 2 konfigurációs lehetőséget biztosítja:

Kapcsolatkonfiguráció tulajdonságai

Ez a szakasz az Azure Service Bushoz való csatlakozáshoz használt konfigurációs beállításokat tartalmazza.

Jegyzet

Ha biztonsági tagot használ egy Azure-erőforrás eléréséhez a Microsoft Entra-azonosítóval történő hitelesítéshez és engedélyezéshez, tekintse meg Hozzáférés engedélyezése a Microsoft Entra-azonosítóval annak biztosításához, hogy a biztonsági tag megkapta-e a megfelelő engedélyt az Azure-erőforrás eléréséhez.

A kapcsolat konfigurálható tulajdonságai spring-cloud-azure-starter-integration-servicebus:

Ingatlan	Típus	Leírás
spring.cloud.azure.servicebus.enabled	logikai érték	Az Azure Service Bus engedélyezése.
spring.cloud.azure.servicebus.connection-string	karakterlánc	A Service Bus-névtér kapcsolati karakterláncának értéke.
spring.cloud.azure.servicebus.custom-endpoint-address	karakterlánc	A Service Bushoz való csatlakozáskor használandó egyéni végpontcím.
spring.cloud.azure.servicebus.namespace	karakterlánc	A Service Bus névtér értéke, amely az FQDN előtagja. Az FQDN-nek a NamespaceName.DomainName elemekből kell állnia
spring.cloud.azure.servicebus.domain-name	karakterlánc	Egy Azure Service Bus-névtér értékének tartományneve.

A Service Bus processzor konfigurációs tulajdonságai

A ServiceBusInboundChannelAdapter a ServiceBusProcessorClient használja az üzenetek felhasználására, egy ServiceBusProcessorClientáltalános tulajdonságainak konfigurálására, a fejlesztők pedig ServiceBusContainerProperties használhatnak a konfigurációhoz. A használatáról lásd a következő szakaszt.

Alapszintű használat

Üzenetek küldése az Azure Service Busba

1. Adja meg a hitelesítő adatok konfigurációs beállításait.

   • A hitelesítő adatok kapcsolati sztringként való megadásához konfigurálja a következő tulajdonságokat a application.yml fájlban:

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

     Jegyzet

     A Microsoft az elérhető legbiztonságosabb hitelesítési folyamat használatát javasolja. Az ebben az eljárásban ismertetett hitelesítési folyamat, például adatbázisok, gyorsítótárak, üzenetkezelés vagy AI-szolgáltatások esetében, nagyon nagy megbízhatóságot igényel az alkalmazásban, és más folyamatokban nem jelenik meg kockázattal. Ezt a folyamatot csak akkor használja, ha a biztonságosabb lehetőségek, például a jelszó nélküli vagy kulcs nélküli kapcsolatokhoz használható felügyelt identitások, nem megvalósíthatók. A helyi gépi műveletekhez előnyben részesítse a jelszó nélküli vagy kulcs nélküli kapcsolatok felhasználói identitásait.

   • A hitelesítő adatok felügyelt identitásként való megadásához konfigurálja a következő tulajdonságokat a application.yml fájlban:

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

Jegyzet

A tenant-id engedélyezett értékek a következők: common, organizations, consumersvagy bérlőazonosító. Ezekről az értékekről további információt az AADSTS50020 hiba – Az identitásszolgáltatótól származó felhasználói fiók nem létezik a bérlőben című cikk Rossz végpont (személyes és szervezeti fiókok) szakaszában talál. Az egybérlős alkalmazás átalakításával kapcsolatos információkért lásd: Egybérlős alkalmazás átalakítása több-bérlős alkalmazássá a Microsoft Entra ID-ban.

• A szolgáltatásnév-azonosítóként használt hitelesítő adatokhoz állítsa be a következő tulajdonságokat a application.yml fájlban:

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

Jegyzet

A tenant-id engedélyezett értékek a következők: common, organizations, consumersvagy bérlőazonosító. Ezekről az értékekről további információt az AADSTS50020 hiba – Az identitásszolgáltatótól származó felhasználói fiók nem létezik a bérlőben című cikk Rossz végpont (személyes és szervezeti fiókok) szakaszában talál. Az egybérlős alkalmazás átalakításával kapcsolatos információkért lásd: Egybérlős alkalmazás átalakítása több-bérlős alkalmazássá a Microsoft Entra ID-ban.

1. Hozzon létre egy DefaultMessageHandler-t a(z) ServiceBusTemplate beannel, hogy üzeneteket küldhessen a Service Busba, és állítsa be a ServiceBusTemplate entitástípusát. Ez a minta a Service Bus-üzenetsort veszi példaként.

   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. Hozzon létre egy üzenetátjáró-kötést a fenti üzenetkezelővel egy üzenetcsatornán keresztül.

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

3. Üzenetek küldése az átjárón keresztül.

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

Üzenetek fogadása az Azure Service Busból

1. Adja meg a hitelesítő adatok konfigurációs beállításait.

2. Definiáljon egy üzenetcsatorna-beant, amely bemeneti csatornaként szolgál.

   @Configuration
   class Demo {
       private static final String INPUT_CHANNEL = "input";
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   }
   

3. A Service Busból érkező üzenetek fogadásához hozza létre a(z) ServiceBusInboundChannelAdapter elemet a ServiceBusMessageListenerContainer bean használatával. Ez a minta a Service Bus-üzenetsor példáját használja.

   @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. Hozzon létre egy üzenet fogadó kötést ServiceBusInboundChannelAdapter a korábban létrehozott üzenetcsatornán keresztül.

   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();
       }
   }
   

A ServiceBusMessageConverter konfigurálása az objectMapper testreszabásához

A ServiceBusMessageConverter konfigurálható bean-ként lett kialakítva, hogy a felhasználók testre szabhassák a ObjectMapper elemet.

A Service Bus üzenetfejlécei

Egyes Service Bus-fejlécek esetében, amelyek több Spring-fejlécállandóra is leképezhetők, a különböző Spring-fejlécek prioritása megjelenik a listában.

Leképezés a Service Bus-fejlécek és a Spring-fejlécek között:

Service Bus-üzenetfejlécek és -tulajdonságok	Tavaszi üzenetfejléc-állandók	Típus	Konfigurálható	Leírás
Tartalomtípus	MessageHeaders#CONTENT_TYPE	karakterlánc	Igen	Az üzenet RFC2045 tartalomtípus-leírója.
Korrelációs azonosító	ServiceBusMessageHeaders#CORRELATION_ID	karakterlánc	Igen	Az üzenet korrelációs azonosítója
Üzenetazonosító	ServiceBusMessageHeaders#MESSAGE_ID	karakterlánc	Igen	Az üzenet üzenetazonosítója; ennek a fejlécnek magasabb a prioritása, mint a MessageHeaders#ID értéké.
Üzenetazonosító	MessageHeaders#ID	UUID (Univerzálisan Egyedi Azonosító)	Igen	Az üzenet üzenetazonosítója; ennek a fejlécnek a prioritása alacsonyabb, mint a ServiceBusMessageHeaders#MESSAGE_ID prioritása.
partíciós kulcs	ServiceBusMessageHeaders#PARTITION_KEY	karakterlánc	Igen	Az üzenet particionált entitásnak való küldéséhez használt partíciókulcs.
Válasz erre:	MessageHeaders#REPLY_CHANNEL	karakterlánc	Igen	Egy entitás címe, amelybe válaszokat szeretne küldeni.
Válasz a munkamenet-azonosítóra	ServiceBusMessageHeaders#REPLY_TO_SESSION_ID	karakterlánc	Igen	Az üzenet ReplyToGroupId tulajdonságértéke.
Ütemezett sorba állítás időpontja (UTC)	ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME	OffsetDateTime	Igen	Az a dátum és időpont, amikor az üzenetet sorba kell állítani a Service Busban; ez a fejléc magasabb prioritású, mint a AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE.
Ütemezett sorba állítás időpontja (UTC)	AzureHeaders#SCHEDULED_ENQUEUE_MESSAGE	Egész szám	Igen	Az az időpont, amikor az üzenetet be kell ütemezni a Service Busba; ennek a fejlécnek kisebb a prioritása, mint a(z) ServiceBusMessageHeaders#SCHEDULED_ENQUEUE_TIME fejlécé.
Munkamenet-azonosító	ServiceBusMessageHeaders#SESSION_ID	karakterlánc	Igen	A munkamenetet kezelő entitás munkamenet-azonosítója.
Az élethez való idő	ServiceBusMessageHeaders#TIME_TO_LIVE	Időtartam	Igen	Az üzenet lejárata előtti időtartam.
Címzett	ServiceBusMessageHeaders#TO	karakterlánc	Igen	Az üzenet „címzett” címe, amely a jövőbeli útválasztási forgatókönyvekhez van fenntartva, és amelyet jelenleg maga a broker figyelmen kívül hagy.
Tárgy	ServiceBusMessageHeaders#SUBJECT	karakterlánc	Igen	Az üzenet tárgya.
Holt betű hiba leírása	ServiceBusMessageHeaders#DEAD_LETTER_ERROR_DESCRIPTION	karakterlánc	Nem	Egy halott betűs üzenet leírása.
A kézbesíthetetlen üzenet oka	ServiceBusMessageHeaders#DEAD_LETTER_REASON	karakterlánc	Nem	Az oka annak, hogy egy üzenet elhalt betűs volt.
Holt betűforrás	ServiceBusMessageHeaders#DEAD_LETTER_SOURCE	karakterlánc	Nem	Az az entitás, amelyben az üzenet holtbetűs lett.
Kézbesítések száma	ServiceBusMessageHeaders#DELIVERY_COUNT	hosszú	Nem	Az üzenet ügyfeleknek való kézbesítésének száma.
Sorba állított sorszám	ServiceBusMessageHeaders#ENQUEUED_SEQUENCE_NUMBER	hosszú	Nem	A Service Bus által az üzenethez rendelt, sorba állításkori sorszám.
Várólista idő	ServiceBusMessageHeaders#ENQUEUED_TIME	OffsetDateTime	Nem	Az az időpont, amikor ez az üzenet várólistára került a Service Busban.
A lejárat dátuma:	ServiceBusMessageHeaders#EXPIRES_AT	OffsetDateTime	Nem	Az az időpont, amikor az üzenet lejár.
Zárolási token	ServiceBusMessageHeaders#LOCK_TOKEN	karakterlánc	Nem	Az aktuális üzenet zárolási tokenje.
Zárolva, amíg	ServiceBusMessageHeaders#LOCKED_UNTIL	OffsetDateTime	Nem	Az az időpont, amikor az üzenet zárolása lejár.
Sorozatszáma	ServiceBusMessageHeaders#SEQUENCE_NUMBER	hosszú	Nem	A Service Bus által egy üzenethez rendelt egyedi szám.
Állam	ServiceBusMessageHeaders#STATE	ServiceBusMessageState	Nem	Az üzenet állapota, amely lehet aktív, késleltetett vagy ütemezett.

Partíciókulcs támogatása

Ez a kezdőmodul támogatja a Service Bus-particionálást azáltal, hogy az üzenetfejlécben lehetővé teszi a partíciókulcs és a munkamenet-azonosító beállítását. Ez a szakasz bemutatja, hogyan állíthat be partíciókulcsot az üzenetekhez.

Ajánlott: Használja a ServiceBusMessageHeaders.PARTITION_KEY a fejléc kulcsaként.

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!");
    }
}

Nem ajánlott, de jelenleg támogatott: AzureHeaders.PARTITION_KEY a fejléc kulcsaként.

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!");
    }
}

Jegyzet

Ha ServiceBusMessageHeaders.PARTITION_KEY és AzureHeaders.PARTITION_KEY is be van állítva az üzenetfejlécekben, ServiceBusMessageHeaders.PARTITION_KEY előnyben részesítjük.

Munkamenet-támogatás

Ez a példa bemutatja, hogyan állíthatja be manuálisan egy üzenet munkamenet-azonosítóját az alkalmazásban.

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!");
    }
}

Jegyzet

Ha a ServiceBusMessageHeaders.SESSION_ID be van állítva az üzenetfejlécekben, és egy másik ServiceBusMessageHeaders.PARTITION_KEY fejléc is be van állítva, a munkamenet-azonosító értéke idővel felülírja a partíciókulcs értékét.

A Service Bus-ügyfél tulajdonságainak testreszabása

A fejlesztők a AzureServiceClientBuilderCustomizer segítségével testre szabhatják a Service Bus-ügyfél tulajdonságait. Az alábbi példa a sessionIdleTimeoutServiceBusClientBuilder tulajdonságát szabja testre:

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

Minták

További információt a azure-spring-boot-samples GitHub adattárában talál.

Spring integráció az Azure Storage Queue-val

Főbb fogalmak

Az Azure Queue Storage nagy mennyiségű üzenet tárolására szolgáló szolgáltatás. A világ bármely pontjáról elérheti az üzeneteket hitelesített hívásokon keresztül HTTP vagy HTTPS használatával. Az üzenetsor-üzenetek mérete legfeljebb 64 KB lehet. Az üzenetsorok több millió üzenetet tartalmazhatnak, akár a tárfiók teljes kapacitáskorlátját is. A sorokat gyakran használják aszinkron módon feldolgozandó feladatlista létrehozására.

Függőségek beállítása

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

Konfiguráció

Ez az alapindító a következő konfigurációs beállításokat biztosítja:

Kapcsolatkonfiguráció tulajdonságai

Ez a szakasz az Azure Storage Queuehoz való csatlakozáshoz használt konfigurációs beállításokat tartalmazza.

Jegyzet

Ha biztonsági tagot használ egy Azure-erőforrás eléréséhez a Microsoft Entra-azonosítóval történő hitelesítéshez és engedélyezéshez, tekintse meg Hozzáférés engedélyezése a Microsoft Entra-azonosítóval annak biztosításához, hogy a biztonsági tag megkapta-e a megfelelő engedélyt az Azure-erőforrás eléréséhez.

A kapcsolat konfigurálható tulajdonságai spring-cloud-azure-starter-integration-storage-queue:

Ingatlan	Típus	Leírás
spring.cloud.azure.storage.queue.enabled	logikai érték	Hogy engedélyezve van-e egy Azure Storage-üzenetsor.
spring.cloud.azure.storage.queue.connection-string	karakterlánc	A Storage Queue Namespace kapcsolati karakterláncának értéke.
spring.cloud.azure.storage.queue.accountName	Karakterlánc	A tárolási várólista-fiók neve.
spring.cloud.azure.storage.queue.accountKey	Karakterlánc	Storage-üzenetsor fiókkulcsa.
spring.cloud.azure.storage.queue.endpoint	karakterlánc	Storage Queue szolgáltatásvégpont.
spring.cloud.azure.storage.queue.sasToken	karakterlánc	SAS-token hitelesítő adatok
spring.cloud.azure.storage.queue.serviceVersion	QueueServiceVersion	Api-kérések készítésekor használt QueueServiceVersion.
spring.cloud.azure.storage.queue.messageEncoding	karakterlánc	Az üzenetsorban lévő üzenet kódolása.

Alapszintű használat

Üzenetek küldése az Azure Storage-üzenetsorba

1. Adja meg a hitelesítő adatok konfigurációs beállításait.

   • A hitelesítő adatok kapcsolati sztringként való megadásához konfigurálja a következő tulajdonságokat a application.yml fájlban:

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

     Megjegyzés

     A Microsoft az elérhető legbiztonságosabb hitelesítési folyamat használatát javasolja. Az ebben az eljárásban ismertetett hitelesítési folyamat, például adatbázisok, gyorsítótárak, üzenetkezelés vagy AI-szolgáltatások esetében, nagyon nagy megbízhatóságot igényel az alkalmazásban, és más folyamatokban nem jelenik meg kockázattal. Ezt a folyamatot csak akkor használja, ha a biztonságosabb lehetőségek, például a jelszó nélküli vagy kulcs nélküli kapcsolatokhoz használható felügyelt identitások, nem megvalósíthatók. A helyi gépi műveletekhez előnyben részesítse a jelszó nélküli vagy kulcs nélküli kapcsolatok felhasználói identitásait.

   • A hitelesítő adatok felügyelt identitásként való megadásához konfigurálja a következő tulajdonságokat a application.yml fájlban:

     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}
     

Megjegyzés

A tenant-id engedélyezett értékek a következők: common, organizations, consumersvagy bérlőazonosító. Ezekről az értékekről további információért lásd a AADSTS50020 hiba – Az identitásszolgáltató felhasználói fiókja nem létezik a bérlőben című cikk Nem megfelelő végpont használata (személyes és szervezeti fiókok esetén) szakaszát. Az egybérlős alkalmazás átalakításával kapcsolatos további információkért lásd: Egybérlős alkalmazás átalakítása több-bérlős alkalmazássá a Microsoft Entra ID-ben.

• A szolgáltatásnév-azonosítóként használt hitelesítő adatokhoz állítsa be a következő tulajdonságokat a application.yml fájlban:

  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}
  

Jegyzet

A tenant-id engedélyezett értékek a következők: common, organizations, consumersvagy bérlőazonosító. Ezekről az értékekről további információt az AADSTS50020 hiba – Az identitásszolgáltatótól származó felhasználói fiók nem létezik a bérlőben című cikk Rossz végpont (személyes és szervezeti fiókok) szakaszában talál. Az egybérlős alkalmazás átalakításával kapcsolatos információkért lásd: Egybérlős alkalmazás átalakítása több-bérlős alkalmazássá a Microsoft Entra ID-ban.

1. Hozza létre DefaultMessageHandler a(z) StorageQueueTemplate bean használatával, hogy üzeneteket küldhessen a Storage Queue-ba.

   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. Hozzon létre egy Üzenetátjáró-kötést a fenti üzenetkezelővel egy üzenetcsatornán keresztül.

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

3. Üzenetek küldése az átjárón keresztül.

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

Üzenetek fogadása az Azure Storage-üzenetsorból

1. Adja meg a hitelesítő adatok konfigurációs beállításait.

2. Definiáljon egy üzenetcsatorna-beant, amely bemeneti csatornaként szolgál.

   class Demo {
       private static final String INPUT_CHANNEL = "input";
   
       @Bean
       public MessageChannel input() {
           return new DirectChannel();
       }
   }
   

3. Hozza létre a(z) StorageQueueMessageSource elemet a(z) StorageQueueTemplate bean használatával, hogy üzeneteket fogadjon a Storage Queue-ból.

   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. Hozzon létre egy üzenet fogadó kötést a StorageQueueMessageSource használatával, amelyet az előző lépésben hoztunk létre a korábban létrehozott üzenetcsatornán keresztül.

   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();
       }
   }
   

Minták

További információt a azure-spring-boot-samples GitHub adattárában talál.