处理消息后关闭kafka使用者
我正在使用@kafkalistener(topics=“${topic}”）来使用来自spring-boot应用程序中某个主题的消息，我需要定期运行该应用程序。spring-kafka版本是2.2.4.发行版。
public class Application {
public static void main(String[] args) throws InterruptedException {
ConfigurableApplicationContext run = SpringApplication.run(Application.class, args);
Thread.sleep(30000);
run.close();
}
}
共有1个答案
看这个答案。
当所有容器实例都空闲时，关闭应用程序。
-
鉴于以下情况： 我在本地启动zookeeper和单个kafka代理，并创建“测试”主题，如kafka快速入门中所述：https://kafka.apache.org/quickstart 然后，我运行一个简单的java程序，该程序每秒向“测试”主题生成一条消息。一段时间后，我关闭了本地的kafka代理，看到制作人继续生成消息，它没有抛出任何异常。最后，我再次启动kafka broker，produ
-
使用kafka流处理器api 场景：流处理器（使用kafka流处理器api实现）从源主题读取数据，并基于某些业务逻辑将数据写入目标主题。
-
我们在kafka中使用Ktabke进行聚合，它非常基本的用途，并参考了kafka文件。 使用Kafka的Streams API处理坏消息的KStream参考 我的用例非常简单，对于任何类型的异常，只需移到错误主题并移到不同的消息
-
如果每个Kafka消息属于一个特定的会话，如何管理会话关联，以便同一个Spark执行器看到链接到一个会话的所有消息？ 如何确保属于会话的消息被Spark executor按照在Kafka中报告的顺序处理？我们能以某种方式实现这一点而不对线程计数施加限制并导致处理开销（如按消息时间戳排序）吗？ 何时检查会话状态？在执行器节点崩溃的情况下，如何从最后一个检查点恢复状态？在驱动程序节点崩溃的情况下，如何
-
我正在使用NSeriveBus构建一个系统，它应该只在特定的时间段将消息发送给远程处理程序。到目前为止，我设法将所有消息放在一个处理队列上，并从那里检查远程处理程序的可用性，如果处理程序不可用，我就不会通过边界发送消息。要做到这一点，我正在使用 但国家安全局会继续努力。这不像其他侦听器将在几分钟内启动并运行，但它可能有几个小时的停机窗口，因此这并不完全有效。 想知道是否有办法让总线稍后重试消息，或
-
我为RabbitMQ制作了一个消费者，作为一个用C#.NET编写的控制台应用程序。它被编程为永久监听队列，每当它在队列中发现消息时，它就处理它。使用者平均每秒处理35条消息。使用者被安排在系统启动时在任务计划程序中运行。消费者运行良好的3-4天。但是，它们继续运行，但不处理任何消息，尽管队列中有消息。当使用者停止并再次启动时，它再次开始正确处理消息。但是，当您手动重新启动时，数以百万计的消息排在队