Spring Cloud Stream Dispatcher 没有订阅者答案

【问题标题】：Spring Cloud Stream Dispatcher has no subscribersSpring Cloud Stream Dispatcher 没有订阅者
【发布时间】：2017-08-23 08:53:16
【问题描述】：

Spring Cloud Stream Dispatcher 没有订阅者错误。

在 Spring Boot 容器成功启动后，我们需要在 Kafka 主题上放置一些通知消息，并且我们的几个微服务执行相同的功能，因此我们编写了一个包含输出通道定义和调度工具的通用 jar。只要我们在 SpringApplication.run 调用之后立即调用 util，该功能就可以按预期工作。

以下是我们的微服务应用程序类示例之一。

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        ConfigurableApplicationContext context =SpringApplication.run(Application.class, args);
        context.getBean(SchedulerConsumerUtils.class).registerOrRestartConsumerJobs();
    }
}

上述设置按预期工作，但这会给开发人员带来不必要的负担，让他们在每个微服务上编写锅炉模板代码。因此，为了避免这种情况，我们编写了一个 Aspect 实现来执行相同的功能，但是使用我们的方面方法，我们遇到了以下错误。

org.springframework.context.ApplicationContextException: 无法启动 bean 'outputBindingLifecycle';嵌套异常是 org.springframework.messaging.MessageDeliveryException：Dispatcher 没有频道“schedulertestsvcs:dev:1180.scheduledJobExecutionResponseOutput”的订阅者。嵌套异常是 org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers

我们尝试了几种方法，例如 Spring SmartLifeCycle 来处理所有 Kafka 输出/输入通道启动完成，但它们都遇到了相同的错误。

以下是我们在 org.springframework.boot.SpringApplication.run(..) 上的 Aspect 实现

@Aspect
@Component
public class SchedulerConsumerAspect {

    @Autowired
    protected ApplicationContext applicationContext;
    @AfterReturning(value = "execution(* org.springframework.boot.SpringApplication.run(..))",returning = "result")
    public void afterConsumerApplicationStartup(JoinPoint pjp, Object result) throws Throwable {
        if(result!=null){
            ConfigurableApplicationContext context=(ConfigurableApplicationContext) result;
            if(context.containsBean("schedulerConsumerUtils")){
                //For what ever reason the following call resulting in Dispatcher has no subscribers for channel error.
                //TODO fix the above issue and enable the following call.
                context.getBean(SchedulerConsumerUtils.class).registerOrRestartConsumerJobs();
            }
        }
    }

}

在我们的调试会话中，我们发现 org.springframework.boot.SpringApplication.run(..) Aspect 在引导过程中被多次调用。首先，当调用方面时，我们得到的结果值为 null，一段时间后 spring boot 调用相同的方面，这次结果不为 null。即使在结果不为 null 之后，组件也没有完全初始化，这就是为什么您会看到对 context.containsBean("schedulerConsumerUtils") 的检查。然而，在 bean 初始化之后，我们看到输出通道没有完全绑定。

处理 Spring Cloud Stream Kafka 输出/输入通道绑定完成的最佳方法是什么？

为什么组件调用在 SpringBoot Application 中可以正常工作，但不能通过 Aspect？我在这几天苦苦挣扎，找不到正确的解决方案。非常感谢任何帮助。

【问题讨论】：

标签： spring-boot binding stream cloud apache-kafka

【解决方案1】：

我遵循了这篇文章Spring cloud stream - send message after application initalization 的建议，并使用了第三个选项 ApplicationRunner。前两个选项对我不起作用。

@Component
public class AppStartup implements ApplicationRunner {
    @Autowired
    protected ApplicationContext applicationContext;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        if(applicationContext!=null){
            applicationContext.getBean(SchedulerConsumerUtils.class).registerOrRestartConsumerJobs();
        }
    }
}

【讨论】：