@KafkaListener 正常关闭，批处理 Kakfa 侦听器不工作

Question

2020-05-14 19：32：11.238 信息 19880 --- [on(4(-127.0.0.1] o.s.c.support.DefaultLifecycle Processor ： 在 30000 的超时时间内，无法关闭 1 个相位值为 2147483547 的 bean： [org.springframework.kafka.config.internalKafkaListenerEndpointRegistry]

Answer 1

默认情况下，Spring 要求每个SmartLifecycle阶段的 bean 在 30 秒内停止;您可以通过添加以下 bean 来更改该行为：

@Bean
public DefaultLifecycleProcessor lifecycleProcessor() {
DefaultLifecycleProcessor lp = new DefaultLifecycleProcessor();
lp.setTimeoutPerShutdownPhase(120_000);
return lp;
}

编辑

在容器的侦听器处理批处理时停止容器不会影响批处理的处理;容器不会终止侦听器线程;但是，默认情况下，容器将在 10 秒后发布容器停止事件(容器属性shutDownTimeout(，即使侦听器实际上尚未完成。

如果您担心生命周期处理器在批处理中途终止线程，并且不想增加其超时，则可以结合使用暂停使用者和侦听事件来执行正常关闭。

下面是一个示例：

@SpringBootApplication
public class So61799727Application {
public static void main(String[] args) {
SpringApplication.run(So61799727Application.class, args);
}

@KafkaListener(id = "so61799727", topics = "so61799727", concurrency = "3")
public void listen(List<String> in) {
System.out.println(in);
}
@Bean
public NewTopic topic() {
return TopicBuilder.name("so61799727").partitions(3).replicas(1).build();
}
@Bean
public ApplicationRunner runner(KafkaTemplate<String, String> template,
KafkaListenerEndpointRegistry registry) {
return args -> {
sendTen(template);
System.out.println("Hit enter to pause container");
System.in.read();
registry.getListenerContainer("so61799727").pause();
};
}
public static void sendTen(KafkaTemplate<String, String> template) {
IntStream.range(0, 10).forEach(i -> template.send("so61799727", "foo" + i));
}
}

@Component
class Eventer {
privaate final KafkaTemplate<String, String> template;
private final AtomicInteger paused = new AtomicInteger();
Eventer(KafkaTemplate<String, String> template) {
this.template = template;
}
@EventListener
public void paused(ConsumerPausedEvent event) {
System.out.println(event);
if (this.paused.incrementAndGet() ==
event.getContainer(ConcurrentMessageListenerContainer.class).getConcurrency()) {
System.out.println("All containers paused");
So61799727Application.sendTen(this.template);
}
}
@EventListener
public void idle(ListenerContainerIdleEvent event) {
System.out.println(event);
}
}

spring.kafka.listener.idle-event-interval=5000
spring.kafka.listener.type=batch
spring.kafka.producer.properties.linger.ms=50