Kafka消费者线程无限订阅 - ReactiveKafkaConsumerTemplate

要解决Kafka消费者线程无限订阅的问题，可以使用ReactiveKafkaConsumerTemplate类来管理消费者线程，并设置订阅的最大持续时间。下面是一个示例代码：

import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import reactor.core.Disposable;

public class KafkaConsumerExample {
    
    private ReactiveKafkaConsumerTemplate<String, String> consumerTemplate;

    public KafkaConsumerExample(ReactiveKafkaConsumerTemplate<String, String> consumerTemplate) {
        this.consumerTemplate = consumerTemplate;
    }

    public void consumeMessages(String topic, long maxDuration) {
        Disposable disposable = consumerTemplate.receiveAutoAck()
                .filter(record -> record.topic().equals(topic))
                .doOnNext(record -> {
                    // 处理接收到的消息
                    System.out.println("Received message: " + record.value());
                })
                .take(Duration.ofSeconds(maxDuration))
                .subscribe();

        // 等待消费者线程完成订阅的最大持续时间后取消订阅
        try {
            Thread.sleep(maxDuration * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        disposable.dispose();
    }

    public static void main(String[] args) {
        // 创建ReactiveKafkaConsumerTemplate实例
        ReactiveKafkaConsumerTemplate<String, String> consumerTemplate = new ReactiveKafkaConsumerTemplate<>();
        KafkaConsumerExample consumerExample = new KafkaConsumerExample(consumerTemplate);

        // 消费指定topic的消息，最大持续时间为10秒
        consumerExample.consumeMessages("my-topic", 10);
    }
}

在上面的示例中，我们首先创建了一个ReactiveKafkaConsumerTemplate实例，并将其传递给KafkaConsumerExample类的构造函数。

在consumeMessages方法中，我们使用consumerTemplate.receiveAutoAck()方法来创建一个消息流，并使用filter方法来过滤出指定的topic的消息。然后，我们使用doOnNext方法来处理接收到的消息，可以根据需要进行自定义的处理逻辑。

接下来，我们使用take方法来限制订阅的最大持续时间，并使用subscribe方法进行订阅。订阅后，消费者线程将开始接收并处理消息。

最后，我们使用Thread.sleep方法在指定的最大持续时间后取消订阅，并调用Disposable.dispose方法来释放资源。

在main方法中，我们创建了一个KafkaConsumerExample实例，并调用consumeMessages方法来消费指定topic的消息，最大持续时间为10秒。

请注意，上面的代码示例中的消费者线程是同步的，即会阻塞当前线程，直到达到最大持续时间。如果需要异步非阻塞的消费者线程，可以使用其他异步编程模型，例如使用Reactor的Mono或Flux。

此外，还需要根据具体的项目和框架，对ReactiveKafkaConsumerTemplate进行适当的配置，例如设置Kafka服务器地址、消费者组ID等。