java - 如何完成kafka消费者安全？(在shutdownHook里面调用thread#join有什么意思吗？)

Question

我正在阅读 this article

这里是完成消费者线程的代码:

Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                System.out.println("Starting exit...");
                consumer.wakeup(); 1
                try {
                    mainThread.join();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

据我了解，ShutdownHook 在所有非守护线程都已完成但进程被操作系统销毁之前调用。
1. 从我的角度来看 mainThread.join() 是无用的。主线程总是会在 ShutdownHook 执行的那一刻结束。它是正确的还是我误解了什么？
2.其实我不明白为什么我们需要等待main thred？我们需要等待关闭方法执行吗？

附言

书中提供了以下主要方法代码:

try {
            // looping until ctrl-c, the shutdown hook will cleanup on exit
            while (true) {
                ConsumerRecords<String, String> records =
                  movingAvg.consumer.poll(1000);
                System.out.println(System.currentTimeMillis() + "--  waiting for data...");
                for (ConsumerRecord<String, String> record :
                  records) {
                    System.out.printf("offset = %d, key = %s,
                      value = %s\n",
                      record.offset(), record.key(),
                      record.value());
                }
                for (TopicPartition tp: consumer.assignment())
                    System.out.println("Committing offset at position:" + consumer.position(tp));
                movingAvg.consumer.commitSync();
            }
        } catch (WakeupException e) {
            // ignore for shutdown 2
        } finally {
            consumer.close(); 3
            System.out.println("Closed consumer and we are done");
        }
    }

Answer 1

您执行 consumer.wakeup() 来中断当前消费者的操作(可能长时间运行(例如轮询)或什至被阻止(在 beginningOffsets(. ..).

将 mainThread.join() 放在那里是为了确保主线程真正完成并且不会在唤醒后的处理过程中关闭。请记住，shutdownHook 还负责处理中断，而不仅仅是普通的程序关闭。

因此，如果您用例如Ctrl-C:

1. shutdown hook gets called
2. main thread is still running, most often waiting for data in `poll`
3. shutdown hook `wakeup`-s the main thread
4. main thread enters the exception handler, breaks the loop
5. main thread closes the consumer with `.close()`
6. shutdown hook waits for 5. and finishes

如果不等待，您可能没有执行步骤 4 和 5 中的消费者关闭步骤。