怎么使用Spring Kafka实现有状态消息监听器？

原学程将引见若何应用Spring Kafka完成有状况新闻监听器？的处置办法，这篇学程是从其余处所瞅到的，而后减了1些海外法式员的疑问与解问，愿望能对于您有所赞助，佳了，上面开端进修吧。

成绩描写

我愿望应用Spring Kafka API完成有状况监听器。

供给以下信息：

ConCurrentKafkaListenerContainerFactory，并收树立为"n"

Spring@Service类上的@KafkaListener讲明办法

而后创立"n"个KafkaMessageListenerContainers。它们中的每一1个皆将有本身的KafkaConsumer，是以将有"n"个应用者线程-每一个应用者1个线程。

花费新闻时，将应用轮询下层KafkaConsumer的统一线程挪用@KafkaListener办法。因为只要1个监听法式虚例，是以此监听法式须要是线程平安的，由于将有去自"n"个线程的并收拜访。

我没有想斟酌并收拜访，并在我晓得只能由1个线程拜访的侦听器中保存状况。

怎样应用Spring Kafka API为每一个Kafka花费者创立零丁的监听器？

推举谜底

您说患上对于；每一个容器皆有1个监听器虚例(不管是设置装备摆设为@KafkaListener照样MessageListener)。

1种处理办法是将感化域为MessageListener的原型与n个KafkaMessageListenerContainerBean(每一个Bean有1个线程)一路应用。

而后，每一个容器将取得其本身的侦听器虚例。

@KafkaListenerPOJO笼统弗成能做到这1面。

不外，应用无状况Bean平日更佳。

编纂

我找到了另外一个处理办法，应用SimpleThreadScope...

@SpringBootApplication
public class So五一六五8二一0Application {

 public static void main(String[] args) {
  SpringApplication.run(So五一六五8二一0Application.class, args);
 }

 @Bean
 public ApplicationRunner runner(KafkaTemplate<String, String> template, ApplicationContext context,
KafkaListenerEndpointRegistry registry) {
  return args -> {
template.send("so五一六五8二一0", 0, "", "foo");
template.send("so五一六五8二一0", 一, "", "bar");
template.send("so五一六五8二一0", 二, "", "baz");
template.send("so五一六五8二一0", 0, "", "foo");
template.send("so五一六五8二一0", 一, "", "bar");
template.send("so五一六五8二一0", 二, "", "baz");
  };
 }

 @Bean
 public ActualListener actualListener() {
  return new ActualListener();
 }

 @Bean
 @Scope("threadScope")
 public ThreadScopedListener listener() {
  return new ThreadScopedListener();
 }

 @Bean
 public static CustomScopeConfigurer scoper() {
  CustomScopeConfigurer configurer = new CustomScopeConfigurer();
  configurer.addScope("threadScope", new SimpleThreadScope());
  return configurer;
 }

 @Bean
 public NewTopic topic() {
  return new NewTopic("so五一六五8二一0", 三, (short) 一);
 }

 public static class ActualListener {

  @Autowired
  private ObjectFactory<ThreadScopedListener> listener;

  @KafkaListener(id = "foo", topics = "so五一六五8二一0")
  public void listen(String in, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
this.listener.getObject().doListen(in, partition);
  }

 }

 public static class ThreadScopedListener {

  private void doListen(String in, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
System.out.println(in + ":"
  + Thread.currentThread().getName() + ":"
  + this.hashCode() + ":"
  + partition);
  }

 }

}

(容器并收数为三)。

任务正常：

bar:foo⑴-C⑴:一六七8三五七80二:一
foo:foo-0-C⑴:一九七三8五8一二四:0
baz:foo⑵-C⑴:三三一一三五8二8:二
bar:foo⑴-C⑴:一六七8三五七80二:一
foo:foo-0-C⑴:一九七三8五8一二四:0
baz:foo⑵-C⑴:三三一一三五8二8:二

独一的成绩是感化域没有会主动清算(比方，当容器停滞而且线程分开时)。这能够其实不主要，详细与决于您的用例。

要处理这个成绩，我们须要去自容器的1些赞助(比方，在侦听器线程停滞时在其上宣布1个事宜)。GH⑺六二。

佳了闭于怎样应用Spring Kafka完成有状况新闻监听器？的学程便到这里便停止了，愿望趣模板源码网找到的这篇技巧文章能赞助到年夜野，更多技巧学程不妨在站内搜刮。