春云流Kafka binder初始化时的通知

Question

我的Spring云流应用程序中有一个简单的Kafka生成器。当我的Spring应用程序启动时，我有一个@PostConstruct方法，它执行一些协调并尝试将事件发送给Kafka生产者。

问题是，当对账开始将enet发送到其中时，我的Kafka制作人还没有准备好，导致以下情况：

org . spring framework . messaging . messagedeliveryexception:调度程序没有通道“orderbook-service-1.orderbook”的订户。；嵌套异常为org . spring framework . integration . messagedispatchingexception:调度程序没有订户，failedMessage=GenericMessage..在org . spring framework . integration . channel . abstractsubscribablechannel . dosend(abstractsubscribablechannel . Java:77)在org . spring framework . integration . channel . abstractmessagechannel . send(abstractmessagechannel . Java:445)

是否有一种方法可以在我的应用程序启动期间获得通知，即Kafka通道已初始化，因此我只启动rec作业发布它。

以下是我的代码片段：

public interface OrderEventChannel {
    String TOPIC_BINDING = "orderbook";
    @Output(TOPIC_BINDING)
    SubscribableChannel outboundEvent();
}

@Configuration
@EnableBinding({OrderEventChannel.class})
@ConditionalOnExpression("${aix.core.stream.outgoing.kafka.enabled:false}")
public class OutgoingKafkaConfiguration {
}

@Service
public class OutgoingOrderKafkaProducer {

    @Autowired
    private OrderEventChannel orderEventChannel;

   public void onOrderEvent( ClientEvent clientEvent ) {

        try {
            Message<KafkaEvent> kafkaMsg = mapToKafkaMessage( clientEvent );
            SubscribableChannel subscribableChannel = orderEventChannel.outboundEvent();
            subscribableChannel.send( kafkaMsg );
        } catch ( RuntimeException rte ) {
            log.error( "Error while publishing Kafka event [{}]", clientEvent, rte );
        }
    }
..
..

}

Answer 1

@PostConstruct在上下文生命周期中开始使用bean太早了；它们仍在创建、配置和连接在一起。

您可以使用 ApplicationListener（或 @EventListener）来侦听 ApplicationReadyEvent（请务必将偶数的 applicationContext 与主应用程序上下文进行比较，因为您可能会获得其他事件）。

您还可以实现 SmartLifecycle 并将代码放入 start（） 中;把你的豆子放在后期阶段，这样它就可以在一切连接好后启动。

输出绑定在阶段 Integer.MIN_VALUE 1000 启动，输入绑定在阶段 Integer.MAX_VALUE - 1000 启动。

因此，如果您想在消息开始流动之前做一些事情，请在它们之间使用一个阶段(例如，0，这是默认值)。