Облачная функция Spring Cosuming rabbitmq queue - у Dispatcher нет подписчиков на канал - PullRequest
Новая
       54

Облачная функция Spring Cosuming rabbitmq queue - у Dispatcher нет подписчиков на канал

0 голосов
/ 15 апреля 2020

Дано: 

  cloud:
    stream:
      rabbit:
      bindings:
        inboundApolloLookupVehicleChannel:
          destination: fed.apollo-vehicle-lookup-test
          group: apollo-mngt-group
          consumer:
            missingQueuesFatal: true
            prefetch: 25
            autoBindDlq: true
            maxAttempts: 1
            republishToDlq: true
            requeueRejected: false
            durableSubscription: true
            maxConcurrency: 6              
      function:
        definition: consume

и: 

@SpringBootApplication
@EnableEurekaClient
@EnableBinding(LookupMessageChannel.class)
public class ApolloLookupServiceApplication {

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


public interface LookupMessageChannel {

    String INBOUND_LOOKUP_VEHICLE_CHANNEL = "inboundApolloLookupVehicleChannel";

    @Input(INBOUND_LOOKUP_VEHICLE_CHANNEL)
    SubscribableChannel inboundApolloLookupVehicleChannel();

}

@Service
@MessageEndpoint
@RequiredArgsConstructor
public class ApolloVehicleLookupService {

    private final ApolloVehicleLookUpRepository apolloVehicleLookUpRepository;
    private static final Logger LOGGER = LoggerFactory.getLogger(ApolloVehicleLookupService.class);

    @Bean
    public Consumer<Flux<ApolloVehicleLookUp>> consume() {
        return stream -> stream             
                .flatMap(this.apolloVehicleLookUpRepository::save)
                .subscribe(value -> {
                        LOGGER.info("stored value: " + value.toString());
                });
    }

}

POM: 



    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.6.RELEASE</version>
        <relativePath />
        <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>14</java.version>
        <spring-cloud.version>Hoxton.SR3</spring-cloud.version>
        <os-maven-plugin.version>1.5.0.Final</os-maven-plugin.version>
        <spotify-docker-maven.version>1.2.0</spotify-docker-maven.version>      
        <os.detected.classifier>linux-x86_64</os.detected.classifier>
    </properties>

  <dependencies>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-rsocket</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-bus-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-config</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
        </dependency>
  </dependencies>

Почему я получаю следующее исключение?

[payload = org.springframework.messaging.MessageDeliveryException: У диспетчера нет подписчиков на канал 'apollo-lookup-service-1.inboundApolloLookupVehicleChannel'. ; вложенное исключение - org.springframework.integration.MessageDispatchingException: у Dispatcher нет подписчиков, failMessage = GenericMessage [payload = byte [439]

Я просто хотел получать сообщения из этой очереди.

Любая помощь будет оценена, спасибо.

1 Ответ

0 голосов
/ 15 апреля 2020

Простое исключение означает, что у вашего канала, определенного в LookupMessageChannel, нет подписчиков, и почему с этого момента вы его не определили. Мне также кажется, что вы пытаетесь использовать функциональный подход, в то время как остальная часть вашего потокового приложения использует устаревшие конфигурации, которые практически устарели.

Не поймите меня неправильно, но в вашем приложении на данный момент мало что не так, поэтому сложно определить, что именно вы ищете. , .

Пожалуйста, рассмотрите этот быстрый старт (5 минут вверху), чтобы получить правильный функциональный подход, а затем не стесняйтесь задавать дополнительные вопросы.

...