Блог

Я хочу получать сообщения из следующих 2 строк подключения

Конечная точка=sb://region1.служебная шина.windows.net/;Имя SharedAccessKeyName=abc;Ключ SharedAccessKey=123;Путь к сущности=Конечная точка моего запроса=sb://region2.servicebus.windows.net/;SharedAccessKeyName=def;SharedAccessKey=456;Путь к сущности=мой запрос

Использовать Java API очень просто

 EventHubConsumerAsyncClient client = new EventHubClientBuilder()
                        .connectionString("Endpoint=sb://region1.servicebus.windows.net/;SharedAccessKeyName=abc;SharedAccessKey=123;EntityPath=my-request")
                        .buildAsyncConsumerClient();
 

Однако, как заставить это работать в файле yaml с помощью Spring Cloud Stream (эквивалентно приведенному выше коду Java)? Перепробовал все учебные пособия, найденные в Интернете, и ни одно из них не работает.

 spring:
      cloud:
        stream:
          function:
            definition: consumeRegion1;consumeRegion2
          bindings:
            consumeRegion1-in-0:
              destination: my-request
              binder: eventhub1
            consumeRegion2-in-0:
              destination: my-request
              binder: eventhub2
          binders:
            eventhub1:
              type: eventhub
              default-candidate: false
              environment:
                spring:
                  cloud:
                    azure:
                      eventhub:
                        connection-string: Endpoint=sb://region1.servicebus.windows.net/;SharedAccessKeyName=abc;SharedAccessKey=123;EntityPath=my-request
            eventhub2:
              type: eventhub
              default-candidate: false
              environment:
                spring:
                  cloud:
                    azure:
                      eventhub:
                        connection-string: Endpoint=sb://region2.servicebus.windows.net/;SharedAccessKeyName=def;SharedAccessKey=456;EntityPath=my-request
 

     @Bean
    public Consumer<Message<String>> consumeRegion1() {
        return message -> {
            System.out.printf(message.getPayload());
        };
    }

    @Bean
    public Consumer<Message<String>> consumeRegion2() {
        return message -> {
            System.out.printf(message.getPayload());
        };
    }
 

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

журнал ошибок

 2021-10-14 21:12:26.760  INFO 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2021-10-14 21:12:26.882  INFO 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2021-10-14 21:12:26.884  INFO 1 --- [           main] trationDelegate$BeanPostProcessorChecker : Bean 'integrationDisposableAutoCreatedBeans' of type [org.springframework.integration.config.annotation.Disposables] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2021-10-14 21:12:29.587  WARN 1 --- [           main] a.s.c.a.e.AzureEventHubAutoConfiguration : Can't construct the EventHubConnectionStringProvider, namespace: null, connectionString: null
2021-10-14 21:12:29.611  INFO 1 --- [           main] a.s.c.a.e.AzureEventHubAutoConfiguration : No event hub connection string provided.
2021-10-14 21:12:30.290  INFO 1 --- [           main] c.a.s.i.eventhub.impl.EventHubTemplate   : Started EventHubTemplate with properties: {checkpointConfig=CheckpointConfig{checkpointMode=RECORD, checkpointCount=0, checkpointInterval=null}, startPosition=LATEST}
2021-10-14 21:12:32.934  INFO 1 --- [           main] c.f.c.c.BeanFactoryAwareFunctionRegistry : Can't determine default function definition. Please use 'spring.cloud.function.definition' property to explicitly define it.