使用reactor-kafka使用不同的线程从Kafka中的消费者组读取数据

import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 创建Kafka消费者配置
        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create()
                .bootstrapServers("kafka-broker1:9092,kafka-broker2:9092")
                .groupId("consumer-group1");

        // 创建Kafka消费者
        KafkaReceiver<String, String> kafkaReceiver = KafkaReceiver.create(receiverOptions);

        // 订阅主题
        kafkaReceiver.receive()
                .subscribeOn(Schedulers.parallel())  // 使用并行线程处理消息
                .subscribe(record -> {
                    // 处理接收到的Kafka消息
                    System.out.println("Received message: " + record.value());
                });

        // 启动消费者
        kafkaReceiver
                .doOnConsumer(consumer -> consumer.subscribe(Collections.singleton("topic1")))
                .doOnConsumer(consumer -> consumer.seekToBeginning(consumer.assignment()))
                .subscribe();
    }
}