远程访问运行在 Kubernetes 中的 Kafka

要远程访问在 Kubernetes 中运行的 Kafka，可以使用以下步骤和代码示例：

步骤1：创建一个 Kafka 服务 首先，您需要在 Kubernetes 集群中部署一个 Kafka 服务。您可以使用 Kafka 操作符（Kafka operator）或 Helm Chart 进行部署。确保 Kafka 服务在集群中成功运行。

步骤2：获取 Kafka 的服务地址和端口 使用以下命令获取 Kafka 服务的地址和端口：

kubectl get svc <kafka-service-name> -n <kafka-namespace>

将 <kafka-service-name> 替换为 Kafka 服务的名称，将 <kafka-namespace> 替换为 Kafka 服务所在的命名空间。

步骤3：创建远程访问 Kafka 的代码 使用以下代码示例创建一个远程访问 Kafka 的应用程序。

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.*;

import java.util.Properties;

public class KafkaRemoteAccessExample {

    public static void main(String[] args) {
        String kafkaServers = "<kafka-service-address>:<kafka-service-port>";
        String topic = "<kafka-topic>";

        // Kafka Producer Example
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers);
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, "key", "value");
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("Message sent successfully!");
            } else {
                System.out.println("Failed to send message: " + exception.getMessage());
            }
        });

        producer.close();

        // Kafka Consumer Example
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServers);
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);

        consumer.subscribe(Collections.singleton(topic));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: " + record.value());
            }
        }
    }
}

在代码中，将 <kafka-service-address> 替换为 Kafka 服务的地址，将 <kafka-service-port> 替换为 Kafka 服务的端口，将 <kafka-topic> 替换为要访问的 Kafka 主题。

步骤4：构建和运行应用程序 使用 Java 编译器将代码编译为可执行的 JAR 文件，并在带有 Kafka 远程访问代码的机器上运行该 JAR 文件。

这样，您就可以远程访问在 Kubernetes 中运行的 Kafka 服务了。