关闭卡夫卡消费者

Question

我读了卡夫卡的高层次消费的细节在此link，看到下面的语句 -

在实践中，更常见的模式是使用睡眠无限期 使用关闭挂钩触发清洁关机。

是否有任何这样做的例子或指针可以帮助？

Answer 1

这将是一个无限循环

public void run() { 
    try { 
     consumer.subscribe(topics); 
     while (true) { 
      ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE); 
     //do something 
     } 
    } catch (WakeupException e) { 
     // do nothing we are shutting down 
    } finally { 
     consumer.close(); 
    } 
    } 

    public void shutdown() { 
    consumer.wakeup(); 
    } 
} 

的例子，这将是你关闭挂钩。

@PostConstruct 
    private void init(){ 
     addShutdownHook(); 
    } 

private void addShutdownHook(){ 
    Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() { 

      @Override 
      public void run() { 
       shutdown(); 
      } 
     })); 
    }