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

如何告诉spring kafka消费者在运行时停止消费?

胡利
2023-03-14

我正在使用带有KafkaListener注释的spring kafka v2.5.2。

在运行时,我希望能够向消费者发送停止消费的信号。

我看到了autoStartup参数,但它似乎只对初始化有效,之后无法更改。

我看到了KafkaListenerEndpointRegistry的methode close()。。。

你有什么建议吗?

提前谢谢。

共有1个答案

蓝昊天
2023-03-14

给每个侦听器一个id属性;然后可以使用KafkaListenerEndpointRegistry bean获取侦听器容器的引用,然后可以启动和停止该容器:getListenerContainer(“myId”)。停止()

此处记录了这一点。

 类似资料:
  • 我有一个简单的Kafka设置。生成器正在以较高的速率向单个分区生成具有单个主题的消息。单个使用者正在使用来自此分区的消息。在此过程中,使用者可能会多次暂停处理消息。停顿可以持续几分钟。生产者停止产生消息后,所有排队的消息都将由使用者处理。生产者产生的消息似乎不会立即被消费者看到。我使用的是Kafka0.10.1.0。这里会发生什么?下面是使用消息的代码部分: 代理上的所有配置都保留为kafka默认

  • 我正在使用Spring Kafka consumer,它从主题中获取消息并将其保存到数据库中。如果满足故障条件,例如db不可用,kafka消费者库是否提供重试机制?如果是,是否有方法设置不同的重试间隔,如第1次重试应在5分钟后进行,第2次重试应在30分钟后进行,第3次重试应在1小时后进行等。

  • 在使用Spring Kafka Consumer时,我有时会收到以下错误消息。如代码片段所示,我至少实现了一次语义 1)我的疑问是,我是否错过了来自消费者的任何信息? 2) 我需要处理这个错误吗。由于 org.apache.kafka.clients.consumer.提交失败异常:无法完成偏移提交,因为消费者不是自动分区分配的活动组的一部分;消费者很可能被踢出组。 我的SpringKafka消费

  • 我想这个话题发生了什么...偏移坏了还是我不知道... 有人知道会发生什么吗?谢谢

  • 我有一个简单的Kafka消费者微服务应用程序,它使用来自某个主题的消息,同一个应用程序运行在两个不同的池中。 所以,当消息由制作人生成,而我的应用程序尝试使用来自主题的消息时,它只被一个池中的一个人使用。 如何停止从消费者Kafka读取并发消息。我想在两个池中使用相同的消息。 这种情况下可能的解决方案是什么

  • 我需要使用consume process Product模式来处理Kafka消息,并已使用Kafka事务管理器配置了Spring Kafka侦听器容器,还设置了事务id前缀以启用Kafka事务。我正在使用批处理的ack模式,并试图了解在这种模式下,在事务中何时提交偏移量。文档似乎表明,一旦使用了轮询中的所有记录,ack模式批提交偏移量——在事务上下文中也是这样吗,即每个轮询1个事务? 或者,在使用