使用Spring IntegrKafka消费者暂停启动Spring Boot应用程序
我正在开发一个Spring Boot应用程序，它使用以Kafka主题为源的Spring集成流。我们的集成流程开始使用一个包含带有spring framework . cloud . stream . annotation . input和Output注释的SubscribableChannels的接口。这些被配置为通过spring . Cloud . stream . Kafka . bindings的云配置从Kafka读取。
当应用程序第一次启动时，它立即开始阅读Kafka主题。这是一个问题，因为应用程序需要初始化一些本地的、不可持久的数据库，然后才能开始正确处理传入的Kafka消息。
我们目前正在使用@PostConstruct在Kafka启动之前填充这些内存数据库，但这不是最理想的，因为应用程序不能使用Eureka、Feign等来可靠地找到具有内存数据库最新数据的健康服务。
由于各种原因，架构不能被更改，使得内存数据库被共享或预填充。要知道，当我称它为内存数据库时，我是在简化一些事情，它实际上是另一种服务。
启动Spring Boot应用程序的最佳方式是什么，以便从Kafka读取的集成流以暂停状态启动，并在其他进程完成后可以取消暂停？
共有1个答案
我假设您使用KafkaMessageDrivenChannelAdapter，根据您提到的Spring集成JavaDSL-Kafka.messageDrivenChannelAdapter（），确切地说。那个可以配置为id和autoStartup（false）。因此它不会立即开始使用Kafka主题。每当您准备好使用时，您可以使用提到的id从应用程序上下文中获取它作为生命周期。
或者您可以向控制总线发送适当的消息。
更新
如果您处理Spring Cloud Stream和Kafka Binder，您应该考虑注入一个< code > BindingsEndpoint bean，并对您的绑定名称和< code >状态执行其< code > changeState(@ Selector String name，State state)。已停止。当内存中的数据库准备就绪时，用< code >状态回调它。开始:https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.发布/引用/html single/# _ binding _ visualization _ and _ control
-
我有一个Spring靴Kafka消费者 为了避免重新平衡，我尝试在KafkaContainer上调用pause（）和resume（），但消费者总是在运行 我错过了什么吗？有人能指导我如何正确地达到要求的行为吗？
-
我正在从数据库中提取数据，以检查我是否有可用的系统资源来处理来自KafkaListener的进一步消息。如果我的条件没有满足，那么我希望@KafkaListener暂停，当条件满足时，我希望@KafkaListener恢复。我如何在SpringKafka实现这一点？ 另外，为特定分区暂停消费者有什么缺点吗？
-
我正在使用这个库来实现节点kafka与消费者暂停和恢复方法来处理背压。我已经创建了一个小演示，我可以在其中和，但问题是在后它停止了消费消息。 这是我的代码。 任何人都可以帮助我，我在恢复消费者时做错了什么？当我启动使用者时，它只接收一条消息，并且在恢复后仍然不消耗任何其他消息。
-
现在，我有一个Spring Boot CLI应用程序，当应用程序启动时，它会自动启动Kafka消费者。我的任务是更新提供API的应用程序，允许在特定条件下启动或停止Kafka消费者。所以，我将使用SpringBootStarterWeb创建该API。但我找不到一种方法来手动管理消费过程。我需要的是 在不使用消费者的情况下启动API 关于如何手动管理消费过程的任何建议？ 技术细节： 用于创建侦听器
-
嗨，我是 Android 编程的新手，我一直在看一个关于 Android 游戏编程的 youtube 教程，问题是这个特定的课程被留在了期中，所以游戏还没有完成，所以我尝试自己添加一些东西。在我的新旅程中，我一直在解决一个自己无法解决的问题，所以如果有人可以帮助我，我会在这里尝试。 问题是游戏的视觉部分，我有一个surfaceView和一个TouchEvent事件。当我暂停游戏（方法暂停活动）时出
-
我已经用Apache ActiveMQ和一个简单的应用程序创建了一个JMS代理，该应用程序将消息纳入队列OK。 我想创建另一个简单的应用程序，使用MDP异步出列这些消息。以下是我到目前为止所拥有的一个例子： 现在我大概需要一个main方法，但是如果消息到达队列时监听器会异步调用onMessage方法，我不确定如何编写代码： 谢谢你的帮助。