《Java 消息中间件·可靠消息与消费治理》4.3 扩缩容与优雅停机

consumer.close(CloseOptions.timeout(Duration.ofSeconds(10))
        .withGroupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP));

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

public class GracefulConsumer {

    private static final String TOPIC = "v2c4.shutdown";

    public static void main(String[] args) throws Exception {
        String name = args[0];

        Properties p = new Properties();
        p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:19292");
        p.put(ConsumerConfig.GROUP_ID_CONFIG, "v2c4-graceful");
        p.put(ConsumerConfig.CLIENT_ID_CONFIG, "graceful-" + name);
        p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10");
        p.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000");
        p.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(p);
        AtomicBoolean running = new AtomicBoolean(true);
        Thread mainThread = Thread.currentThread();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println(name + " SHUTDOWN-HOOK: stop flag + wakeup");
            running.set(false);
            consumer.wakeup();                 // 安全地打断 poll()
            try {
                mainThread.join(15000);        // 给主线程留出提交与关闭的时间
            } catch (InterruptedException ignored) {
                Thread.currentThread().interrupt();
            }
        }, "shutdown-" + name));

        try {
            consumer.subscribe(List.of(TOPIC), new ConsumerRebalanceListener() {
                @Override
                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                    // 撤销前提交，缩小下一个持有者的重复窗口。
                    try {
                        consumer.commitSync(Duration.ofSeconds(5));
                    } catch (Exception e) {
                        System.out.println(name + " commit on revoke failed: " + e);
                    }
                    System.out.println(name + " REVOKED " + partitions);
                }

                @Override
                public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                    System.out.println(name + " ASSIGNED " + partitions);
                }
            });

            while (running.get()) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
                for (ConsumerRecord<String, String> r : records) {
                    process(r);                // 业务处理
                }
                if (!records.isEmpty()) {
                    consumer.commitSync();     // 处理完再提交
                }
            }
        } catch (WakeupException e) {
            if (running.get()) {
                throw e;                       // 不是我们发起的 wakeup，向上抛
            }
            System.out.println(name + " WAKEUP received, leaving loop");
        } finally {
            try {
                consumer.commitSync(Duration.ofSeconds(5));
                System.out.println(name + " final commit ok");
            } catch (Exception e) {
                System.out.println(name + " final commit failed: " + e);
            }
            consumer.close(CloseOptions.timeout(Duration.ofSeconds(10))
                    .withGroupMembershipOperation(CloseOptions.GroupMembershipOperation.LEAVE_GROUP));
            System.out.println(name + " closed");
        }
    }

    private static void process(ConsumerRecord<String, String> r) throws Exception {
        Thread.sleep(20);                      // 模拟业务耗时
    }
}

13:21:23.301 [main] INFO ConsumerCoordinator - [Consumer clientId=graceful-A, groupId=v2c4-graceful] Successfully joined group with generation Generation{generationId=1, memberId='graceful-A-9fe12c1a-...', protocol='range'}
A ASSIGNED [v2c4.shutdown-0, v2c4.shutdown-1, v2c4.shutdown-2]
A SHUTDOWN-HOOK: stop flag + wakeup
13:21:24.854 [main] INFO ConsumerRebalanceListenerInvoker - [Consumer clientId=graceful-A, groupId=v2c4-graceful] Revoke previously assigned partitions [v2c4.shutdown-0, v2c4.shutdown-1, v2c4.shutdown-2]
A REVOKED [v2c4.shutdown-0, v2c4.shutdown-1, v2c4.shutdown-2]
13:21:24.859 [main] INFO ConsumerCoordinator - [Consumer clientId=graceful-A, groupId=v2c4-graceful] Member graceful-A-9fe12c1a-... sending LeaveGroup request to coordinator 127.0.0.1:19292 ... due to the consumer is being closed
A final commit ok
A closed, processed=60

13:21:26.611 [main] INFO ConsumerRebalanceListenerInvoker - [Consumer clientId=graceful-B, groupId=v2c4-graceful] Revoke previously assigned partitions [v2c4.shutdown-3, v2c4.shutdown-4, v2c4.shutdown-5]
13:21:26.638 [main] INFO ConsumerCoordinator - [Consumer clientId=graceful-B, groupId=v2c4-graceful] Successfully joined group with generation Generation{generationId=2, ...}
B ASSIGNED [v2c4.shutdown-0, v2c4.shutdown-1, v2c4.shutdown-2, v2c4.shutdown-3, v2c4.shutdown-4, v2c4.shutdown-5]

logs/seq_A.txt   60 行
logs/seq_B.txt 1140 行
合计 1200 次处理
distinct=1200
processed_times=1 records=1200
seq 范围 1..1200

13:22:44.287 [kafka-coordinator-heartbeat-thread | v2c4-stall] WARN ConsumerCoordinator -
[Consumer clientId=stall-A, groupId=v2c4-stall] consumer poll timeout has expired. This means the time
between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically
implies that the poll loop is spending too much time processing messages. You can address this either by
increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with
max.poll.records.
13:22:44.288 [kafka-coordinator-heartbeat-thread | v2c4-stall] INFO ConsumerCoordinator -
Member stall-A-... sending LeaveGroup request to coordinator 127.0.0.1:19292 ... due to consumer poll timeout has expired.

总投递尝试 1212 次
distinct=1200
processed_times=2 records=10
processed_times=3 records=1
A closed processed=16
B closed processed=1196