Kafka消费者心跳线程在Java Nio上阻塞

问题描述：Kafka消费者心跳线程在Java Nio上阻塞，导致消费者无法正常工作。

解决方法：

1. 使用单独的线程处理Kafka消费者心跳的逻辑，避免阻塞主线程。

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {

    public static void main(String[] args) {
        // 创建Kafka消费者配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 创建Kafka消费者
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("test-topic"));

        // 创建心跳线程
        Thread heartbeatThread = new Thread(() -> {
            while (true) {
                // 发送心跳请求
                consumer.poll(Duration.ZERO);
            }
        });

        // 启动心跳线程
        heartbeatThread.start();

        // 开始消费消息
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    // 处理消息逻辑
                    System.out.println("Received message: " + record.value());
                }
            }
        } finally {
            consumer.close();
            // 停止心跳线程
            heartbeatThread.interrupt();
            try {
                heartbeatThread.join();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

2. 使用Kafka的异步提交偏移量功能，避免在消费者心跳线程中阻塞。

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.WakeupException;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class KafkaConsumerExample {

    private static final Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

    public static void main(String[] args) {
        // 创建Kafka消费者配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 创建Kafka消费者
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅主题
        consumer.subscribe(Collections.singletonList("test-topic"));

        // 创建心跳线程
        Thread heartbeatThread = new Thread(() -> {
            try {
                while (true) {
                    // 发送心跳请求
                    consumer.poll(Duration.ZERO);
                    // 提交偏移量
                    consumer.commitAsync(offsets, null);
                }
            } catch (WakeupException e) {
                // 忽略WakeupException异常
            } finally {
                // 提交偏移量
                consumer.commitSync(offsets);
                consumer.close();
            }
        });

        // 启动心跳线程
        heartbeatThread.start();

        // 开始消费消息
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    // 处理消息逻辑
                    System.out.println("Received message: " + record.value());

                    // 更新偏移量
                    offsets.put(new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(record.offset() + 1));
                }
            }
        } finally {
            // 停止心跳线程
            heartbeatThread.interrupt();
            try {
                heartbeatThread.join();
            } catch (InterruptedException e