当前位置: 首页 > 知识库问答 >
问题:

使用@KafkaListener注释时如何暂停Kafka消费者

武峻熙
2023-03-14

我正在从数据库中提取数据,以检查我是否有可用的系统资源来处理来自KafkaListener的进一步消息。如果我的条件没有满足,那么我希望@KafkaListener暂停,当条件满足时,我希望@KafkaListener恢复。我如何在SpringKafka实现这一点?

另外,为特定分区暂停消费者有什么缺点吗?

共有1个答案

傅增
2023-03-14

当前版本中没有pause/resume控件。只有当框架检测到慢速侦听器时才会自动执行

在最新的2.0中,我们遇到了相反的情况:我们不再提供自动暂停/恢复——没有任何理由,但同时允许使用消费者参数配置@KafkaListener方法,这样您就可以在那里执行任何所需的逻辑。

 类似资料:
  • 我已经实现了Kafka消费者,现在我有了一个场景。 从Kafka流2.2.5中读取数据。通过Srpingboot发布 加载数据库表1 将数据从表1复制到表2 清理桌子1 要执行上述操作,我需要使用quartz的调度作业(已编写)暂停/恢复Kafka使用者,该作业将数据从表1复制到表2。但是在这个活动中,我希望我的Kafka听众暂停,一旦复制完成,它应该继续。 我的实施:

  • 我想用C#听Kafka主题的消息。 与Java中一样,还有一个注释@KafkaListener,当添加到函数上方时,它会侦听来自主题的消息,然后执行函数的逻辑。 示例-@KafkaListener(topics=“topicname”,groupId=“groupId”)//这里的函数代码 同样,C#中是否有使用Confluent的注释。Kafka

  • 我想在特定时间停止对特定主题的轮询。 Spring防尘套2.X Springkafka 2.5.5 Kafka版本2.5.1 比如即使有消息进来测试题目分区,消息也是从00到01堆在分区里,没有消耗。 01点之后,我想再次使用有关TEST主题的消息。 如何暂停和恢复?

  • 我正在使用这个库来实现节点kafka与消费者暂停和恢复方法来处理背压。我已经创建了一个小演示,我可以在其中和,但问题是在后它停止了消费消息。 这是我的代码。 任何人都可以帮助我,我在恢复消费者时做错了什么?当我启动使用者时,它只接收一条消息,并且在恢复后仍然不消耗任何其他消息。

  • 我正在使用spring boot 2.1.7.Release和spring-kafka 2.2.7.Release.并且我正在使用@KafKalistener注释来创建一个消费者,并且我正在使用消费者的所有默认设置。 根据apache kafka文档,'max.poll.records'的默认值是500。 在这里,我试图理解spring是如何处理记录处理的。现在我的问题是,如果我们已经在主题a上发

  • 我正在使用Brave库https://github.com/openzipkin/brave进行跟踪,现在我也想将其用于Kafka消费者。我想避免添加Spring Sleuth,并利用Brave Kafka仪器https://github.com/openzipkin/brave/tree/master/instrumentation/kafka-clients. 对于Kafka消费者,我使用@K