Spring Cloud Stream에 대한 Spring Cloud Azure 지원

Spring Cloud Stream은 공유 메시징 시스템과 연결된 확장성이 뛰어난 이벤트 기반 마이크로 서비스를 빌드하기 위한 프레임워크입니다.

프레임워크는 이미 확립되고 친숙한 Spring 관용구 및 모범 사례를 기반으로 구축된 유연한 프로그래밍 모델을 제공합니다. 이러한 모범 사례에는 영구 게시/하위 의미 체계, 소비자 그룹 및 상태 저장 파티션에 대한 지원이 포함됩니다.

현재 바인더 구현은 다음과 같습니다.

• spring-cloud-azure-stream-binder-eventhubs - 자세한 내용은 Azure Event Hubs Spring Cloud Stream Binder 참조하세요.
• - 자세한 내용은 Azure Service Bus Spring Cloud Stream Binder를 참조하세요.

Azure Event Hubs용 Spring Cloud Stream Binder

주요 개념

Azure Event Hubs용 Spring Cloud Stream Binder는 Spring Cloud Stream 프레임워크에 대한 바인딩 구현을 제공합니다. 이 구현에서는 Spring Integration Event Hubs 채널 어댑터를 기초로 사용합니다. 디자인의 관점에서 Event Hubs는 Kafka와 유사합니다. 또한 Kafka API를 통해 Event Hubs에 액세스할 수 있습니다. 프로젝트에 Kafka API에 대한 종속성이 엄격한 경우 Kafka API 샘플 사용하여 이벤트 허브를 수 있습니다.

소비자 그룹

Event Hubs는 Apache Kafka와 유사한 소비자 그룹을 지원하지만 논리는 약간 다릅니다. Kafka는 모든 커밋된 오프셋을 broker에 저장하지만 수동으로 처리되는 Event Hubs 메시지의 오프셋을 저장해야 합니다. Event Hubs SDK는 Azure Storage 내에 이러한 오프셋을 저장하는 함수를 제공합니다.

분할 지원

Event Hubs는 Kafka와 유사한 물리적 파티션 개념을 제공합니다. 그러나 소비자와 파티션 간의 Kafka 자동 재조정과 달리 Event Hubs는 일종의 선점 모드를 제공합니다. 스토리지 계정은 어떤 소비자가 어떤 파티션을 소유하는지 결정하는 임대 역할을 합니다. 새 소비자가 시작되면 가장 많이 로드된 소비자로부터 일부 파티션을 도용하여 워크로드 균형을 달성하려고 합니다.

부하 분산 전략을 지정하기 위해 spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.load-balancing.* 속성이 제공됩니다. 자세한 내용은 소비자 속성 섹션을 참조하세요.

Batch 소비자 지원

Spring Cloud Azure Stream Event Hubs 바인더는 Spring Cloud Stream Batch Consumer 기능지원합니다.

일괄 처리 소비자 모드를 사용하려면 spring.cloud.stream.bindings.<binding-name>.consumer.batch-mode 속성을 true설정합니다. 사용하도록 설정하면 일괄 처리된 이벤트 목록의 페이로드가 포함된 메시지가 수신되고 Consumer 함수에 전달됩니다. 또한 각 메시지 헤더는 목록으로 변환되며, 그 중 콘텐츠는 각 이벤트에서 구문 분석된 연결된 헤더 값입니다. 파티션 ID, 검사점 및 마지막으로 큐에 추가된 속성의 공동 헤더는 이벤트의 전체 일괄 처리가 동일한 값을 공유하므로 단일 값으로 표시됩니다. 자세한 내용은 Spring IntegrationSpring Cloud Azure 지원의 Event Hubs 메시지 헤더 섹션을 참조하세요.

일괄 처리 소비자의 검사점은 BATCH 및 MANUAL두 가지 모드를 지원합니다. BATCH 모드는 바인더가 이벤트를 받으면 전체 이벤트 일괄 처리의 검사점을 지정하는 자동 검사점 모드입니다. MANUAL 모드는 사용자가 이벤트의 검사점을 지정하는 것입니다. 사용하는 경우 Checkpointer 메시지 헤더에 전달되고 사용자는 이를 사용하여 검사점을 수행할 수 있습니다.

접두사 max-size있는 max-wait-time 및 spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.batch. 속성을 설정하여 일괄 처리 크기를 지정할 수 있습니다. max-size 속성이 필요하며 max-wait-time 속성은 선택 사항입니다. 자세한 내용은 소비자 속성 섹션을 참조하세요.

종속성 설정

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

또는 Maven에 대한 다음 예제와 같이 Spring Cloud Azure Stream Event Hubs Starter를 사용할 수도 있습니다.

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

구성

바인더는 구성 옵션의 다음 세 부분을 제공합니다.

연결 구성 속성

이 섹션에는 Azure Event Hubs에 연결하는 데 사용되는 구성 옵션이 포함되어 있습니다.

spring-cloud-azure-stream-binder-eventhubs의 연결 구성 가능한 속성:

바인더는 기본적으로 Spring Could Azure Resource Manager 지원합니다. Data 관련 역할로 부여되지 않은 보안 주체를 사용하여 연결 문자열을 검색하는 방법에 대한 자세한 내용은 Spring Could Azure Resource ManagerBasic 사용량 섹션을 참조하세요.

검사점 구성 속성

이 섹션에는 파티션 소유권 및 검사점 정보를 유지하는 데 사용되는 Storage Blobs 서비스에 대한 구성 옵션이 포함되어 있습니다.

spring-cloud-azure-stream-binder-eventhubs의 구성 가능한 속성 검사점:

Azure Event Hubs 바인딩 구성 속성

다음 옵션은 소비자 속성, 고급 소비자 구성, 생산자 속성 및 고급 생산자 구성의 네 가지 섹션으로 나뉩니다.

소비자 속성

이러한 속성은 EventHubsConsumerProperties통해 노출됩니다.

spring-cloud-azure-stream-binder-eventhubs의 소비자 구성 가능한 속성:

spring:
  cloud:
    stream:
      eventhubs:
        bindings:
          <binding-name>:
            consumer:
              initial-partition-event-position:
                0:
                  offset: earliest
                1:
                  sequence-number: 100
                2:
                  enqueued-date-time: 2022-01-12T13:32:47.650005Z
                4:
                  inclusive: false

고급 소비자 구성

위의 연결, 검사점및 일반적인 Azure SDK 클라이언트 구성은 접두사 spring.cloud.stream.eventhubs.bindings.<binding-name>.consumer.구성할 수 있는 각 바인더 소비자에 대한 사용자 지정을 지원합니다.

생산자 속성

이러한 속성은 EventHubsProducerProperties통해 노출됩니다.

spring-cloud-azure-stream-binder-eventhubs의 생산자 구성 가능한 속성:

고급 생산자 구성

위의 연결 및 일반적인 Azure SDK 클라이언트 구성은 접두사 spring.cloud.stream.eventhubs.bindings.<binding-name>.producer.사용하여 구성할 수 있는 각 바인더 생산자에 대한 사용자 지정을 지원합니다.

기본 사용량

Event Hubs에서/받는 메시지 보내기 및 받기

1. 구성 옵션을 자격 증명 정보로 채웁니다.

   • 연결 문자열로 자격 증명의 경우 application.yml 파일에서 다음 속성을 구성합니다.

     spring:
       cloud:
         azure:
           eventhubs:
             connection-string: ${EVENTHUB_NAMESPACE_CONNECTION_STRING}
             processor:
               checkpoint-store:
                 container-name: ${CHECKPOINT_CONTAINER}
                 account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                 account-key: ${CHECKPOINT_ACCESS_KEY}
         function:
           definition: consume;supply
         stream:
           bindings:
             consume-in-0:
               destination: ${EVENTHUB_NAME}
               group: ${CONSUMER_GROUP}
             supply-out-0:
               destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
           eventhubs:
             bindings:
               consume-in-0:
                 consumer:
                   checkpoint:
                     mode: MANUAL
     

     메모

     사용 가능한 가장 안전한 인증 흐름을 사용하는 것이 권장됩니다. 데이터베이스, 캐시, 메시징 또는 AI 서비스와 같이 이 절차에서 설명하는 인증 흐름은 애플리케이션에 대한 신뢰 수준이 매우 높고 다른 흐름에 존재하지 않는 위험을 수반합니다. 암호 없는 연결 또는 키 없는 연결에 대한 관리 ID와 같은 더 안전한 옵션이 실행 가능하지 않은 경우에만 이 흐름을 사용합니다. 로컬 컴퓨터 작업의 경우 암호 없는 연결이나 키 없는 연결에 사용자 ID를 사용하는 것이 좋습니다.

   • 서비스 주체 자격 증명의 경우 application.yml 파일에서 다음 속성을 구성합니다.

     spring:
       cloud:
         azure:
           credential:
             client-id: ${AZURE_CLIENT_ID}
             client-secret: ${AZURE_CLIENT_SECRET}
           profile:
             tenant-id: <tenant>
           eventhubs:
             namespace: ${EVENTHUB_NAMESPACE}
             processor:
               checkpoint-store:
                 container-name: ${CONTAINER_NAME}
                 account-name: ${ACCOUNT_NAME}
         function:
           definition: consume;supply
         stream:
           bindings:
             consume-in-0:
               destination: ${EVENTHUB_NAME}
               group: ${CONSUMER_GROUP}
             supply-out-0:
               destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
           eventhubs:
             bindings:
               consume-in-0:
                 consumer:
                   checkpoint:
                     mode: MANUAL
     

• 자격 증명을 관리 ID로 사용하려면 application.yml 파일에서 다음 속성을 구성합니다.

  spring:
    cloud:
      azure:
        credential:
          managed-identity-enabled: true
          client-id: ${AZURE_MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity
        eventhubs:
          namespace: ${EVENTHUB_NAMESPACE}
          processor:
            checkpoint-store:
              container-name: ${CONTAINER_NAME}
              account-name: ${ACCOUNT_NAME}
      function:
        definition: consume;supply
      stream:
        bindings:
          consume-in-0:
            destination: ${EVENTHUB_NAME}
            group: ${CONSUMER_GROUP}
          supply-out-0:
            destination: ${THE_SAME_EVENTHUB_NAME_AS_ABOVE}
  
        eventhubs:
          bindings:
            consume-in-0:
              consumer:
                checkpoint:
                  mode: MANUAL
  

1. 공급업체 및 소비자를 정의합니다.

   @Bean
   public Consumer<Message<String>> consume() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                   message.getPayload(),
                   message.getHeaders().get(EventHubsHeaders.PARTITION_KEY),
                   message.getHeaders().get(EventHubsHeaders.SEQUENCE_NUMBER),
                   message.getHeaders().get(EventHubsHeaders.OFFSET),
                   message.getHeaders().get(EventHubsHeaders.ENQUEUED_TIME)
           );
   
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("Hello world, " + i++).build();
       };
   }
   

분할 지원

보낼 메시지에 대한 파티션 정보를 구성하기 위해 사용자가 제공한 파티션 정보가 포함된 PartitionSupplier 만들어집니다. 다음 순서도는 파티션 ID 및 키에 대해 다른 우선 순위를 가져오는 프로세스를 보여줍니다.

분할 지원 프로세스의 순서도를 보여 주는

Batch 소비자 지원

1. 다음 예제와 같이 일괄 처리 구성 옵션을 제공합니다.

   spring:
     cloud:
       function:
         definition: consume
       stream:
         bindings:
           consume-in-0:
             destination: ${AZURE_EVENTHUB_NAME}
             group: ${AZURE_EVENTHUB_CONSUMER_GROUP}
             consumer:
               batch-mode: true
         eventhubs:
           bindings:
             consume-in-0:
               consumer:
                 batch:
                   max-batch-size: 10 # Required for batch-consumer mode
                   max-wait-time: 1m # Optional, the default value is null
                 checkpoint:
                   mode: BATCH # or MANUAL as needed
   

2. 공급업체 및 소비자를 정의합니다.

   BATCH검사점 모드의 경우 다음 코드를 사용하여 메시지를 보내고 일괄 처리로 사용할 수 있습니다.

   @Bean
   public Consumer<Message<List<String>>> consume() {
       return message -> {
           for (int i = 0; i < message.getPayload().size(); i++) {
               LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                       message.getPayload().get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_OFFSET)).get(i),
                       ((List<Object>) message.getHeaders().get(EventHubsHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i));
           }
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("\"test"+ i++ +"\"").build();
       };
   }
   

   MANUAL검사점 모드의 경우 다음 코드를 사용하여 메시지를 보내고 일괄 처리로 사용/검사점을 사용할 수 있습니다.

   @Bean
   public Consumer<Message<List<String>>> consume() {
       return message -> {
           for (int i = 0; i < message.getPayload().size(); i++) {
               LOGGER.info("New message received: '{}', partition key: {}, sequence number: {}, offset: {}, enqueued time: {}",
                   message.getPayload().get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_PARTITION_KEY)).get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_SEQUENCE_NUMBER)).get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_OFFSET)).get(i),
                   ((List<Object>) message.getHeaders().get(EventHubHeaders.BATCH_CONVERTED_ENQUEUED_TIME)).get(i));
           }
   
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           checkpointer.success()
                       .doOnSuccess(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                       .doOnError(error -> LOGGER.error("Exception found", error))
                       .block();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("\"test"+ i++ +"\"").build();
       };
   }
   

오류 메시지 처리

Event Hubs 메시지 헤더

지원되는 기본 메시지 헤더는 Spring Integration대한 Spring Cloud Azure 지원의 Event Hubs 메시지 헤더 섹션을 참조하세요.

여러 바인더 지원

여러 Event Hubs 네임스페이스에 대한 연결도 여러 바인더를 사용하여 지원됩니다. 이 샘플에서는 연결 문자열을 예로 사용합니다. 서비스 주체 및 관리 ID의 자격 증명도 지원됩니다. 각 바인더의 환경 설정에서 관련 속성을 설정할 수 있습니다.

1. Event Hubs에서 여러 바인더를 사용하려면 application.yml 파일에서 다음 속성을 구성합니다.

   spring:
     cloud:
       function:
         definition: consume1;supply1;consume2;supply2
       stream:
         bindings:
           consume1-in-0:
             destination: ${EVENTHUB_NAME_01}
             group: ${CONSUMER_GROUP_01}
           supply1-out-0:
             destination: ${THE_SAME_EVENTHUB_NAME_01_AS_ABOVE}
           consume2-in-0:
             binder: eventhub-2
             destination: ${EVENTHUB_NAME_02}
             group: ${CONSUMER_GROUP_02}
           supply2-out-0:
             binder: eventhub-2
             destination: ${THE_SAME_EVENTHUB_NAME_02_AS_ABOVE}
         binders:
           eventhub-1:
             type: eventhubs
             default-candidate: true
             environment:
               spring:
                 cloud:
                   azure:
                     eventhubs:
                       connection-string: ${EVENTHUB_NAMESPACE_01_CONNECTION_STRING}
                       processor:
                         checkpoint-store:
                           container-name: ${CHECKPOINT_CONTAINER_01}
                           account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                           account-key: ${CHECKPOINT_ACCESS_KEY}
           eventhub-2:
             type: eventhubs
             default-candidate: false
             environment:
               spring:
                 cloud:
                   azure:
                     eventhubs:
                       connection-string: ${EVENTHUB_NAMESPACE_02_CONNECTION_STRING}
                       processor:
                         checkpoint-store:
                           container-name: ${CHECKPOINT_CONTAINER_02}
                           account-name: ${CHECKPOINT_STORAGE_ACCOUNT}
                           account-key: ${CHECKPOINT_ACCESS_KEY}
         eventhubs:
           bindings:
             consume1-in-0:
               consumer:
                 checkpoint:
                   mode: MANUAL
             consume2-in-0:
               consumer:
                 checkpoint:
                   mode: MANUAL
         poller:
           initial-delay: 0
           fixed-delay: 1000
   

   메모

   이전 애플리케이션 파일은 모든 바인딩에 애플리케이션에 대한 단일 기본 폴러를 구성하는 방법을 보여 줍니다. 특정 바인딩에 대해 폴러를 구성하려면 spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000같은 구성을 사용할 수 있습니다.

   메모

   사용 가능한 가장 안전한 인증 흐름을 사용하는 것이 권장됩니다. 데이터베이스, 캐시, 메시징 또는 AI 서비스와 같이 이 절차에서 설명하는 인증 흐름은 애플리케이션에 대한 신뢰 수준이 매우 높고 다른 흐름에 존재하지 않는 위험을 수반합니다. 암호 없는 연결 또는 키 없는 연결에 대한 관리 ID와 같은 더 안전한 옵션이 실행 가능하지 않은 경우에만 이 흐름을 사용합니다. 로컬 컴퓨터 작업의 경우 암호 없는 연결이나 키 없는 연결에 사용자 ID를 사용하는 것이 좋습니다.

2. 두 개의 공급업체와 두 명의 소비자를 정의해야 합니다.

   @Bean
   public Supplier<Message<String>> supply1() {
       return () -> {
           LOGGER.info("Sending message1, sequence1 " + i);
           return MessageBuilder.withPayload("Hello world1, " + i++).build();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply2() {
       return () -> {
           LOGGER.info("Sending message2, sequence2 " + j);
           return MessageBuilder.withPayload("Hello world2, " + j++).build();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume1() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message1 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message1 '{}' successfully checkpointed", message))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume2() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message2 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(success -> LOGGER.info("Message2 '{}' successfully checkpointed", message))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   

리소스 프로비저닝

Event Hubs 바인더는 이벤트 허브 및 소비자 그룹의 프로비저닝을 지원하며, 사용자는 다음 속성을 사용하여 프로비저닝을 사용하도록 설정할 수 있습니다.

spring:
  cloud:
    azure:
      credential:
        tenant-id: <tenant>
      profile:
        subscription-id: ${AZURE_SUBSCRIPTION_ID}
      eventhubs:
        resource:
          resource-group: ${AZURE_EVENTHUBS_RESOURECE_GROUP}

샘플

자세한 내용은 GitHub의 azure-spring-boot-samples 리포지토리를 참조하세요.

Azure Service Bus용 Spring Cloud Stream Binder

주요 개념

Azure Service Bus용 Spring Cloud Stream Binder는 Spring Cloud Stream Framework에 대한 바인딩 구현을 제공합니다. 이 구현에서는 Spring Integration Service Bus 채널 어댑터를 기초로 사용합니다.

예약된 메시지

이 바인더는 지연된 처리를 위해 토픽에 메시지 전송을 지원합니다. 사용자는 헤더 x-delay 메시지의 지연 시간을 밀리초 단위로 표현하여 예약된 메시지를 보낼 수 있습니다. 메시지는 x-delay 밀리초 후에 해당 토픽에 전달됩니다.

소비자 그룹

Service Bus 토픽은 Apache Kafka와 비슷한 소비자 그룹을 지원하지만 논리는 약간 다릅니다. 이 바인더는 소비자 그룹 역할을 하는 토픽의 Subscription 사용합니다.

종속성 설정

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

또는 Maven에 대한 다음 예제와 같이 Spring Cloud Azure Stream Service Bus Starter를 사용할 수도 있습니다.

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

구성

바인더는 구성 옵션의 다음 두 부분을 제공합니다.

연결 구성 속성

이 섹션에는 Azure Service Bus에 연결하는 데 사용되는 구성 옵션이 포함되어 있습니다.

spring-cloud-azure-stream-binder-servicebus의 연결 구성 가능한 속성:

바인더는 기본적으로 Spring Could Azure Resource Manager 지원합니다. Data 관련 역할로 부여되지 않은 보안 주체를 사용하여 연결 문자열을 검색하는 방법에 대한 자세한 내용은 Spring Could Azure Resource ManagerBasic 사용량 섹션을 참조하세요.

Azure Service Bus 바인딩 구성 속성

다음 옵션은 소비자 속성, 고급 소비자 구성, 생산자 속성 및 고급 생산자 구성의 네 가지 섹션으로 나뉩니다.

소비자 속성

이러한 속성은 ServiceBusConsumerProperties통해 노출됩니다.

spring-cloud-azure-stream-binder-servicebus의 소비자 구성 가능한 속성:

고급 소비자 구성

위의 연결 및 일반적인 Azure SDK 클라이언트 구성은 접두사 spring.cloud.stream.servicebus.bindings.<binding-name>.consumer.사용하여 구성할 수 있는 각 바인더 소비자에 대한 사용자 지정을 지원합니다.

생산자 속성

이러한 속성은 ServiceBusProducerProperties통해 노출됩니다.

spring-cloud-azure-stream-binder-servicebus의 생산자 구성 가능한 속성:

고급 생산자 구성

위의 연결 및 일반적인 Azure SDK 클라이언트 구성은 접두사 spring.cloud.stream.servicebus.bindings.<binding-name>.producer.사용하여 구성할 수 있는 각 바인더 생산자에 대한 사용자 지정을 지원합니다.

기본 사용량

Service Bus에서/받는 메시지 보내기 및 받기

1. 구성 옵션을 자격 증명 정보로 채웁니다.

   • 연결 문자열로 자격 증명의 경우 application.yml 파일에서 다음 속성을 구성합니다.

         spring:
           cloud:
             azure:
               servicebus:
                 connection-string: ${SERVICEBUS_NAMESPACE_CONNECTION_STRING}
             function:
               definition: consume;supply
             stream:
               bindings:
                 consume-in-0:
                   destination: ${SERVICEBUS_ENTITY_NAME}
                   # If you use Service Bus Topic, add the following configuration
                   # group: ${SUBSCRIPTION_NAME}
                 supply-out-0:
                   destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
               servicebus:
                 bindings:
                   consume-in-0:
                     consumer:
                       auto-complete: false
                   supply-out-0:
                     producer:
                       entity-type: queue # set as "topic" if you use Service Bus Topic
     

     메모

     사용 가능한 가장 안전한 인증 흐름을 사용하는 것이 권장됩니다. 데이터베이스, 캐시, 메시징 또는 AI 서비스와 같이 이 절차에서 설명하는 인증 흐름은 애플리케이션에 대한 신뢰 수준이 매우 높고 다른 흐름에 존재하지 않는 위험을 수반합니다. 암호 없는 연결 또는 키 없는 연결에 대한 관리 ID와 같은 더 안전한 옵션이 실행 가능하지 않은 경우에만 이 흐름을 사용합니다. 로컬 컴퓨터 작업의 경우 암호 없는 연결이나 키 없는 연결에 사용자 ID를 사용하는 것이 좋습니다.

   • 서비스 주체 자격 증명의 경우 application.yml 파일에서 다음 속성을 구성합니다.

         spring:
           cloud:
             azure:
               credential:
                 client-id: ${AZURE_CLIENT_ID}
                 client-secret: ${AZURE_CLIENT_SECRET}
               profile:
                 tenant-id: <tenant>
               servicebus:
                 namespace: ${SERVICEBUS_NAMESPACE}
             function:
               definition: consume;supply
             stream:
               bindings:
                 consume-in-0:
                   destination: ${SERVICEBUS_ENTITY_NAME}
                   # If you use Service Bus Topic, add the following configuration
                   # group: ${SUBSCRIPTION_NAME}
                 supply-out-0:
                   destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
               servicebus:
                 bindings:
                   consume-in-0:
                     consumer:
                       auto-complete: false
                   supply-out-0:
                     producer:
                       entity-type: queue # set as "topic" if you use Service Bus Topic
     

• 자격 증명을 관리 ID로 사용하려면 application.yml 파일에서 다음 속성을 구성합니다.

      spring:
        cloud:
          azure:
            credential:
              managed-identity-enabled: true
              client-id: ${MANAGED_IDENTITY_CLIENT_ID} # Only needed when using a user-assigned managed identity
            servicebus:
              namespace: ${SERVICEBUS_NAMESPACE}
          function:
            definition: consume;supply
          stream:
            bindings:
              consume-in-0:
                destination: ${SERVICEBUS_ENTITY_NAME}
                # If you use Service Bus Topic, add the following configuration
                # group: ${SUBSCRIPTION_NAME}
              supply-out-0:
                destination: ${SERVICEBUS_ENTITY_NAME_SAME_AS_ABOVE}
            servicebus:
              bindings:
                consume-in-0:
                  consumer:
                    auto-complete: false
                supply-out-0:
                  producer:
                    entity-type: queue # set as "topic" if you use Service Bus Topic
  

1. 공급업체 및 소비자를 정의합니다.

   @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(success -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(error -> LOGGER.error("Exception found", error))
                   .block();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply() {
       return () -> {
           LOGGER.info("Sending message, sequence " + i);
           return MessageBuilder.withPayload("Hello world, " + i++).build();
       };
   }
   

파티션 키 지원

바인더는 메시지 헤더에서 파티션 키 및 세션 ID를 설정하여 Service Bus 분할 지원합니다. 이 섹션에서는 메시지의 파티션 키를 설정하는 방법을 소개합니다.

Spring Cloud Stream은 파티션 키 SpEL 식 속성 spring.cloud.stream.bindings.<binding-name>.producer.partition-key-expression제공합니다. 예를 들어 이 속성을 "'partitionKey-' + headers[<message-header-key>]" 설정하고 message-header-key라는 헤더를 추가합니다. Spring Cloud Stream은 식을 평가할 때 이 헤더의 값을 사용하여 파티션 키를 할당합니다. 다음 코드는 예제 생산자를 제공합니다.

@Bean
public Supplier<Message<String>> generate() {
    return () -> {
        String value = "random payload";
        return MessageBuilder.withPayload(value)
            .setHeader("<message-header-key>", value.length() % 4)
            .build();
    };
}

세션 지원

바인더는 Service Bus의 메시지 세션을 지원합니다. 메시지 헤더를 통해 메시지의 세션 ID를 설정할 수 있습니다.

@Bean
public Supplier<Message<String>> generate() {
    return () -> {
        String value = "random payload";
        return MessageBuilder.withPayload(value)
            .setHeader(ServiceBusMessageHeaders.SESSION_ID, "Customize session ID")
            .build();
    };
}

오류 메시지 처리

Service Bus 메시지 헤더

지원되는 기본 메시지 헤더는 Spring Integration대한 Spring Cloud Azure 지원의 Service Bus 메시지 헤더 섹션을 참조하세요.

여러 바인더 지원

여러 Service Bus 네임스페이스에 대한 연결도 여러 바인더를 사용하여 지원됩니다. 이 샘플에서는 연결 문자열을 예로 사용합니다. 서비스 주체 및 관리 ID의 자격 증명도 지원되며, 사용자는 각 바인더의 환경 설정에서 관련 속성을 설정할 수 있습니다.

1. ServiceBus의 여러 바인더를 사용하려면 application.yml 파일에서 다음 속성을 구성합니다.

   spring:
     cloud:
       function:
         definition: consume1;supply1;consume2;supply2
       stream:
         bindings:
           consume1-in-0:
             destination: ${SERVICEBUS_TOPIC_NAME}
             group: ${SUBSCRIPTION_NAME}
           supply1-out-0:
             destination: ${SERVICEBUS_TOPIC_NAME_SAME_AS_ABOVE}
           consume2-in-0:
             binder: servicebus-2
             destination: ${SERVICEBUS_QUEUE_NAME}
           supply2-out-0:
             binder: servicebus-2
             destination: ${SERVICEBUS_QUEUE_NAME_SAME_AS_ABOVE}
         binders:
           servicebus-1:
             type: servicebus
             default-candidate: true
             environment:
               spring:
                 cloud:
                   azure:
                     servicebus:
                       connection-string: ${SERVICEBUS_NAMESPACE_01_CONNECTION_STRING}
           servicebus-2:
             type: servicebus
             default-candidate: false
             environment:
               spring:
                 cloud:
                   azure:
                     servicebus:
                       connection-string: ${SERVICEBUS_NAMESPACE_02_CONNECTION_STRING}
         servicebus:
           bindings:
             consume1-in-0:
               consumer:
                 auto-complete: false
             supply1-out-0:
               producer:
                 entity-type: topic
             consume2-in-0:
               consumer:
                 auto-complete: false
             supply2-out-0:
               producer:
                 entity-type: queue
         poller:
           initial-delay: 0
           fixed-delay: 1000
   

   메모

   이전 애플리케이션 파일은 모든 바인딩에 애플리케이션에 대한 단일 기본 폴러를 구성하는 방법을 보여 줍니다. 특정 바인딩에 대해 폴러를 구성하려면 spring.cloud.stream.bindings.<binding-name>.producer.poller.fixed-delay=3000같은 구성을 사용할 수 있습니다.

   메모

   사용 가능한 가장 안전한 인증 흐름을 사용하는 것이 권장됩니다. 데이터베이스, 캐시, 메시징 또는 AI 서비스와 같이 이 절차에서 설명하는 인증 흐름은 애플리케이션에 대한 신뢰 수준이 매우 높고 다른 흐름에 존재하지 않는 위험을 수반합니다. 암호 없는 연결 또는 키 없는 연결에 대한 관리 ID와 같은 더 안전한 옵션이 실행 가능하지 않은 경우에만 이 흐름을 사용합니다. 로컬 컴퓨터 작업의 경우 암호 없는 연결이나 키 없는 연결에 사용자 ID를 사용하는 것이 좋습니다.

2. 공급업체 2개와 소비자 2개를 정의해야 합니다.

   @Bean
   public Supplier<Message<String>> supply1() {
       return () -> {
           LOGGER.info("Sending message1, sequence1 " + i);
           return MessageBuilder.withPayload("Hello world1, " + i++).build();
       };
   }
   
   @Bean
   public Supplier<Message<String>> supply2() {
       return () -> {
           LOGGER.info("Sending message2, sequence2 " + j);
           return MessageBuilder.withPayload("Hello world2, " + j++).build();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume1() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message1 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(e -> LOGGER.error("Error found", e))
                   .block();
       };
   }
   
   @Bean
   public Consumer<Message<String>> consume2() {
       return message -> {
           Checkpointer checkpointer = (Checkpointer) message.getHeaders().get(CHECKPOINTER);
           LOGGER.info("New message2 received: '{}'", message);
           checkpointer.success()
                   .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message.getPayload()))
                   .doOnError(e -> LOGGER.error("Error found", e))
                   .block();
       };
   
   }
   

리소스 프로비저닝

Service Bus 바인더는 큐, 토픽 및 구독의 프로비저닝을 지원하며, 사용자는 다음 속성을 사용하여 프로비저닝을 사용하도록 설정할 수 있습니다.

spring:
  cloud:
    azure:
      credential:
        tenant-id: <tenant>
      profile:
        subscription-id: ${AZURE_SUBSCRIPTION_ID}
      servicebus:
        resource:
          resource-group: ${AZURE_SERVICEBUS_RESOURECE_GROUP}
    stream:
      servicebus:
        bindings:
          <binding-name>:
            consumer:
              entity-type: ${SERVICEBUS_CONSUMER_ENTITY_TYPE}

Service Bus 클라이언트 속성 사용자 지정

개발자는 AzureServiceClientBuilderCustomizer 사용하여 Service Bus 클라이언트 속성을 사용자 지정할 수 있습니다. 다음 예제에서는 sessionIdleTimeoutServiceBusClientBuilder 속성을 사용자 지정합니다.

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

샘플

자세한 내용은 GitHub의 azure-spring-boot-samples 리포지토리를 참조하세요.