Spring Cloud Stream-First Kafka消息得到错误“Dispatcher没有订阅者”

Question

我的应用程序成功发送Kafka消息，但只有在Kafka初始化后。在此之前，我得到错误“Dispatcher没有订阅者”。如何等待订阅者完成频道注册？

• 17.165 SenderClass已创建
• 17.816初始化类，@PostConstruct启动PollingTask
• 24.781PollingTask发送第一条Kafka消息
• 24.816第一个错误：“Dispatcher没有订阅服务器”
• 25.778注册MessageChannel my-channel
• 仍然看到调度程序错误
• 27.067频道my-channel'有1个订阅者
• 此后不再出错，消息发送正常

我不知道该怎么做。疯狂的猜测包括：

1. 将发送代码放在@PostConstruct
中
2. 向发件人添加@AutoConfigureBefore(BindingServiceConfiguration.class)
3. 向SenderClass添加@AutoConfigureAfter（BindingServiceConfiguration.class）
4. 将@AutoConfigureBefore(BindingServiceConfiguration.class)添加到main
5. 将@Dependson（{“EnableBindingClass”}）放置在任务上
6. 将@Dependson（{“ApplicationLifeCycle”}）放置在ScheduerClass上，其中ApplicationLifeCycle是一个类，它除了实现SmartLifecycle外什么也不做，并且getPhase返回max_int
7. 确保整个包的ComponentScan处于打开状态（来自其他SO线程的建议）
8. 以上各种组合

public interface Source {
  @Output(channelName)
  MessageChannel outboundChannel();
}

@EnableBinding(Source.class)
@Component
public class Sender {
  @Autowired
  private Source source;

  public boolean send(SomeObject object) {
    return source.outboundChannel().send(MessageBuilder.withPayload(object).build());
  }

@Service
public class Scheduler {
  @Autowired
  Sender sender;
  @Autowired
  ThreadPoolTaskScheduler taskScheduler;

  @PostConstruct
  public void initialize() {
    taskScheduler.schedule(new PollingTask(), nextTime);
  }

  private class PollingTask implements Runnable {
    @Override
    public void run() {
      List<SomeObject> objects = getDummyData();
      for(SomeObject object : objects)
      {
        sender.send(interval);
      }

      Instant nextTime = Instant.now().plusMillis(1_000L);
      try {
        taskScheduler.schedule(new PollingTask(), nextTime);
      } catch (Exception e) {
        logger.error(e);
      }
    }
}

现在管用了！在启动发送消息的事情的调度程序中，我从@PostConstruct中的启动事情切换到了SmartLifecycle：：start（）。

@Service
public class Scheduler implements SmartLifecycle {
  @Autowired
  Sender sender;
  @Autowired
  ThreadPoolTaskScheduler taskScheduler;

  @Override
  public void start() {
    taskScheduler.schedule(new PollingTask(), nextTime);
  }

  private class PollingTask implements Runnable {
    @Override
    public void run() {
      List<SomeObject> objects = getDummyData();
      for(SomeObject object : objects)
      {
        sender.send(interval);
      }

      Instant nextTime = Instant.now().plusMillis(1_000L);
      try {
        taskScheduler.schedule(new PollingTask(), nextTime);
      } catch (Exception e) {
        logger.error(e);
      }
    }
}

Answer 1

@PostConstruct发送消息为时过早；上下文仍在构建中。IMPLEMT SmartLifecycle，将bean置于高阶段(integer.max_value)，并在start（）中执行发送操作。

或者在ApplicationRunner中执行发送操作。