如何确保Spring云流监听器在启动时等待处理消息，直到应用程序完全初始化？

Question

使用Spring Cloud Stream Kafka应用程序，我们如何确保流侦听器等待处理消息，直到一些依赖任务（例如引用数据填充）完成？下面的应用程序无法处理消息，因为消息传递得太早。我们如何保证Spring Boot App中的这种排序？

@Service
public class ApplicationStartupService implements ApplicationRunner {

  private final FooReferenceDataService fooReferenceDataService;

  @Override
  public void run(ApplicationArguments args) throws Exception {
      fooReferenceDataService.loadData();
  }
}

@EnableBinding(MyBinding.class)
public class MyFooStreamProcessor {

  @Autowired FooService fooService;

  @StreamListener("my-input")
  public void process(KStream<String, Foo> input) {
      input.foreach((k,v)-> {
          // !!! this fails to save
          // messages are delivered too early before foo reference data got loaded into database
          fooService.save(v);
      });         
  }
}

< li >春-云-流:2.1.0.RELEASE < li >Spring启动:2.1.2 .释放

我发现截至 2018 年 5 月 15 日，Spring云流中不可用。

Kafka -延迟绑定，直到复杂的服务初始化完成

我们是否有一个计划/时间表来支持这一点？

Answer 1

与此同时，我通过使用@Ordered和ApplicationRunner实现了我想要的目标。它很乱，但有效。基本上，流侦听器将等到其他工作完成。

@Service
@Order(1)
public class ApplicationStartupService implements ApplicationRunner {

  private final FooReferenceDataService fooReferenceDataService;

  @Override
  public void run(ApplicationArguments args) throws Exception {
      fooReferenceDataService.loadData();
  }
}

@EnableBinding(MyBinding.class)
@Order(2)
public class MyFooStreamProcessor implements ApplicationRunner {

  @Autowired FooService fooService;
  private final AtomicBoolean ready = new AtomicBoolean(false);

  @StreamListener("my-input")
  public void process(KStream<String, Foo> input) {
      input.foreach((k,v)-> {
          while (ready.get() == false) {
            try {
              log.info("sleeping for other dependent components to finish initialization");
              Thread.sleep(10000);
            } catch (InterruptedException e) {
              log.info("woke up");
            }
          }
          fooService.save(v);
      });         
  }

  @Override
  public void run(ApplicationArguments args) throws Exception {
    ready.set(true);
  }
}