java - Spring Cloud函数合并RabbitMQ队列-调度程序没有 channel 的订阅者

鉴于：

  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：调度程序没有订阅者，failedMessage = GenericMessage [payload = byte [439]

我只想使用该队列中的消息。

任何帮助，将不胜感激，谢谢。

最佳答案

简单的异常意味着您在LookupMessageChannel中定义的频道没有订阅者，为什么要这么做，因为您尚未定义任何订阅者。在我看来，您还尝试使用功能性方法，而流应用程序的其余部分使用的旧配置几乎已弃用。

别误会我的意思，但是您的应用目前有几处错，因此很难确定您到底想做什么。 。 。

请考虑遵循this quick start（持续5分钟），以使您掌握正确的功能方法，然后随时提出其他问题。

关于java - Spring Cloud函数合并RabbitMQ队列-调度程序没有 channel 的订阅者，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/61222082/