Spring Cloud Stream Azure Service Bus

Ez a cikk bemutatja, hogyan küldhet és fogadhat üzeneteket a Spring Cloud Stream Binderrel Azure Service Bus queues és topics Spring Boot-alkalmazásokban.

Azure Azure Service Bus ("Service Bus") nevű aszinkron üzenetkezelési platformot biztosít, amely a Advanced Message Queueing Protocol 1.0 ("AMQP 1.0") szabványon alapul. A Service Bus számos támogatott Azure-platformon használható.

Előfeltételek

Feljegyzés

Ha hozzáférést szeretne adni a fiókjának a Azure Service Bus erőforrásaihoz, rendelje hozzá a Azure Service Bus Data Sender szerepköröket az Azure Service Bus Data Receiver éppen használt Microsoft Entra fiókhoz. A hozzáférési szerepkörök engedélyezésével kapcsolatos további információkért lásd: Az Azure szerepkörök hozzárendelése a Azure portálon és A Azure Service Bus entitások eléréséhez Microsoft Entra ID rendelkező alkalmazások hitelesítése és engedélyezése.

Fontos

A cikkben ismertetett lépések elvégzéséhez a Spring Boot 2.5-ös vagy újabb verziójára van szükség.

Üzenetek küldése és fogadása Azure Service Bus

A Azure Service Bus üzenetsorával vagy témakörével üzeneteket küldhet és fogadhat a Spring Cloud Azure Stream Binder Service Bus használatával.

A Spring Cloud Azure Stream Binder Service Bus modul telepítéséhez adja hozzá a következő függőségeket a pom.xml fájlhoz:

  • A Spring Cloud Azure Anyagjegyzék (BOM):

    <dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>com.azure.spring</groupId>
      <artifactId>spring-cloud-azure-dependencies</artifactId>
      <version>7.4.0</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

    Feljegyzés

    Ha Spring Boot 4.0.x-et használ, állítsa a verziót a spring-cloud-azure-dependencies következőre 7.4.0: .

    Ha a Spring Boot 3.5.x-et használja, állítsa a verziót a spring-cloud-azure-dependencies következőre 6.5.0: .

    Ha a Spring Boot 3.1.x-3.5.x verziót használja, mindenképpen állítsa be a verziót spring-cloud-azure-dependencies-ra/c1.

    Ha a Spring Boot 2.x-et használja, mindenképpen állítsa be a spring-cloud-azure-dependencies verziót 4.20.0-re.

    Ezt az anyagjegyzéket (BOM) a <dependencyManagement> fájl szakaszában kell konfigurálni. Ez biztosítja, hogy minden Spring Cloud-Azure függőség ugyanazt a verziót használja.

    Az anyagjegyzékhez használt verzióval kapcsolatos további információkért lásd: Melyik verzióját a Spring Cloud Azure-nak érdemes használnom?.

  • A Spring Cloud Azure Stream Binder Service Bus összetevő:

    <dependency>
     <groupId>com.azure.spring</groupId>
     <artifactId>spring-cloud-azure-stream-binder-servicebus</artifactId>
</dependency>

Az alkalmazás kódolása

Az alábbi lépésekkel konfigurálhatja az alkalmazást úgy, hogy Service Bus üzenetsort vagy témakört használjon üzenetek küldéséhez és fogadásához.

  1. Konfigurálja a Service Bus hitelesítő adatokat a konfigurációs fájlban application.properties.

     spring.cloud.azure.servicebus.namespace=${AZURE_SERVICEBUS_NAMESPACE}
 spring.cloud.stream.bindings.consume-in-0.destination=${AZURE_SERVICEBUS_QUEUE_NAME}
 spring.cloud.stream.bindings.supply-out-0.destination=${AZURE_SERVICEBUS_QUEUE_NAME}
 spring.cloud.stream.servicebus.bindings.consume-in-0.consumer.auto-complete=false
 spring.cloud.stream.servicebus.bindings.supply-out-0.producer.entity-type=queue
 spring.cloud.function.definition=consume;supply;
 spring.cloud.stream.poller.fixed-delay=60000 
 spring.cloud.stream.poller.initial-delay=0

    Az alábbi táblázat a konfiguráció mezőit ismerteti:

    Mező Leírás
    spring.cloud.azure.servicebus.namespace Adja meg a névteret, amelyet a Service Bus-hoz az Azure portálon szerzett be.
    spring.cloud.stream.bindings.consume-in-0.destination Adja meg a jelen oktatóanyagban használt Service Bus üzenetsort vagy témakört.
    spring.cloud.stream.bindings.supply-out-0.destination Adja meg ugyanazt az értéket, amelyet a bemeneti célként használt.
    spring.cloud.stream.servicebus.bindings.consume-in-0.consumer.auto-complete Adja meg, hogy automatikusan rendezze-e az üzeneteket. Ha falseként van beállítva, a rendszer hozzáadja a Checkpointer üzenetfejlécét, hogy a fejlesztők manuálisan rendezhessék az üzeneteket.
    spring.cloud.stream.servicebus.bindings.supply-out-0.producer.entity-type Adja meg a kimeneti kötés entitástípusát, lehet queue vagy topic.
    spring.cloud.function.definition Adja meg, hogy melyik funkcionális komponenst kell a kötések által meghatározott külső célhelyhez kapcsolni.
    spring.cloud.stream.poller.fixed-delay Adja meg az alapértelmezett poller rögzített késleltetését ezredmásodpercben. Az alapértelmezett érték a 1000 L. A javasolt érték a következő 60000: .
    spring.cloud.stream.poller.initial-delay Adja meg a rendszeres eseményindítók kezdeti késleltetését. Az alapértelmezett érték a 0.

  2. Szerkessze az indítási osztályfájlt a következő tartalom megjelenítéséhez.

    import com.azure.spring.messaging.checkpoint.Checkpointer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.function.Consumer;
import java.util.function.Supplier;
import static com.azure.spring.messaging.AzureHeaders.CHECKPOINTER;

@SpringBootApplication
public class ServiceBusQueueBinderApplication implements CommandLineRunner {

    private static final Logger LOGGER = LoggerFactory.getLogger(ServiceBusQueueBinderApplication.class);
    private static final Sinks.Many<Message<String>> many = Sinks.many().unicast().onBackpressureBuffer();

    public static void main(String[] args) {
        SpringApplication.run(ServiceBusQueueBinderApplication.class, args);
    }

    @Bean
    public Supplier<Flux<Message<String>>> supply() {
        return ()->many.asFlux()
                       .doOnNext(m->LOGGER.info("Manually sending message {}", m))
                       .doOnError(t->LOGGER.error("Error encountered", t));
    }

    @Bean
    public Consumer<Message<String>> consume() {
        return message->{
            Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
            LOGGER.info("New message received: '{}'", message.getPayload());
            checkpointer.success()
                        .doOnSuccess(s->LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                        .doOnError(e->LOGGER.error("Error found", e))
                        .block();
        };
    }

    @Override
    public void run(String... args) {
        LOGGER.info("Going to add message {} to Sinks.Many.", "Hello World");
        many.emitNext(MessageBuilder.withPayload("Hello World").build(), Sinks.EmitFailureHandler.FAIL_FAST);
    }

}

    Tipp.

    Ebben az oktatóanyagban nincsenek hitelesítési műveletek a konfigurációkban vagy a kódban. Az Azure szolgáltatásokhoz való csatlakozáshoz azonban hitelesítés szükséges. A hitelesítés befejezéséhez Azure Identitást kell használnia. A Spring Cloud Azure DefaultAzureCredential használ, amelyet az Azure Identitástár biztosít a hitelesítő adatok kódmódosítások nélküli beszerzéséhez.

    DefaultAzureCredential több hitelesítési módszert támogat, és meghatározza, hogy melyik metódust használja futásidőben. Ez a megközelítés lehetővé teszi, hogy az alkalmazás különböző hitelesítési módszereket használjon különböző környezetekben (például helyi és éles környezetekben) környezetspecifikus kód implementálása nélkül. További információ: DefaultAzureCredential.

    A helyi fejlesztési környezetekben a hitelesítés befejezéséhez használhat Azure CLI, Visual Studio Code, PowerShellt vagy más módszereket. További információ: Azure hitelesítés Java fejlesztési környezetekben. Ha Azure üzemeltetési környezetekben szeretné elvégezni a hitelesítést, javasoljuk, hogy használja a felhasználó által hozzárendelt felügyelt identitást. További információ: Mi az Azure-erőforrások felügyelt identitásai?

  3. Indítsa el az alkalmazást. Az alábbi példához hasonló üzenetek az alkalmazásnaplóban jelennek meg:

    New message received: 'Hello World'
Message 'Hello World' successfully checkpointed

Következő lépések

Azure Spring-fejlesztőknekFelhő Azure Stream Binder Service Bus-minták

  • Last updated on