#选择连接地址及配置连接方式

业务应用访问 Kafka 实例时，有多种连接方式。本文档介绍如何选择连接地址，通过该地址访问 Kafka 实例，并提供示例。

#服务器提供的连接方式说明及适用场景

创建 Kafka 实例时，服务器会提供多种访问链接组合，并附带相关参数说明：

• 内外网访问：应用与 Kafka 服务器是否处于同一业务集群内。
• 认证机制：是否开启认证以验证客户端身份。认证机制支持 SCRAM-SHA-512 和 TLS 协议（注：TLS 认证机制仅支持在 TLS 加密传输链路上启用）。
• 传输加密：保障数据传输过程安全，采用数据加密技术防止数据在网络传输中被截获或窃听。传输加密协议通常为 TLS。

#客户端配置说明

连接时，客户端需根据以下信息选择合适的链接并配置对应参数：

以下示例展示几种典型访问链接（Java 客户端），其他组合可根据合适链接进行配置：

• 内网访问，无认证，非加密传输
• 内网访问，有认证（TLS 协议），加密传输
• 外网访问，有认证（SCRAM-SHA-512 协议），非加密传输
• 外网访问，有认证（SCRAM-SHA-512 协议），加密传输
• 外网访问，有认证（TLS 协议），加密传输

#1. 内网访问 - 无认证，非加密传输

此时客户端与 Kafka 实例位于同一业务集群内。Kafka 服务器未开启认证，传输过程无需加密。客户端只需获取 Kafka 实例的内网路由地址。

可复制【Kafka 实例 - 访问方式】标签页中“集群内访问 - 明文传输”的连接地址，替换到实际代码片段中。

客户端连接代码示例如下：

package com.lanzhiwang.producer;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class ExampleProducer {
    public static void main(String[] args) {

        Properties kafkaProps = new Properties();
        kafkaProps.put("bootstrap.servers", "test-kafka-bootstrap:9092");
        kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // ProducerRecord(String topic, K key, V value)
        ProducerRecord<String, String> record = new ProducerRecord<>("my-cluster-topic", "Precision Products", "China");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps)) {
            // 忽略返回值，无法判断消息是否发送成功。
            producer.send(record);
            // Thread.sleep(2000000); // 延迟 2000 秒
        } catch (Exception e) {
            // 生产者发送消息到 Kafka 前遇到错误。
            e.printStackTrace();
        }
    }
}

#2. 内网访问 - 有认证（TLS 协议），加密传输

此时客户端与 Kafka 实例位于同一业务集群内。Kafka 服务器开启认证，传输过程需加密。客户端需获取 Kafka 实例的内网路由地址、TLS 协议的用户密码证书及传输证书。

1. Kafka 实例内网路由地址

   可复制【Kafka 实例 - 访问方式】标签页中“集群内访问 - TLS 传输”的连接地址，替换到实际代码片段中。

2. TLS 协议传输及密码证书

   使用 kubectl 命令生成用户及密码证书，并将相关证书复制到客户端对应路径。具体命令如下：

   • 生成 CA 证书：

     $ kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12

   • 生成 CA 证书密码：

     $ kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d

3. TLS 协议用户及密码证书

   使用 kubectl 命令生成传输及密码证书，并将相关证书复制到客户端对应路径。具体命令如下：

   • 生成用户证书：

     $ kubectl -n <namespace> get secret <Kafka User Name> -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12

   • 生成用户证书密码：

     $ kubectl -n <namespace> get secret <Kafka User Name> -o jsonpath='{.data.user\.password}' | base64 -d

4. 客户端连接代码示例如下：

   package com.lanzhiwang.producer;
   import java.util.Properties;
   
   import org.apache.kafka.clients.producer.KafkaProducer;
   import org.apache.kafka.clients.producer.ProducerRecord;
   
   public class ExampleProducer {
       public static void main(String[] args) {
   
           Properties kafkaProps = new Properties();
           kafkaProps.put("bootstrap.servers", "test-kafka-bootstrap:9093");
           kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
           kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
   
           // 配置 TLS/SSL 属性
           kafkaProps.put("security.protocol", "SSL");
           // kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12
           kafkaProps.put("ssl.truststore.location", "/Users/temp/certs/ca.p12");
           // kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d
           kafkaProps.put("ssl.truststore.password", "************");
           // kubectl -n <namespace> get secret <Kafka User Name> -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12
           kafkaProps.put("ssl.keystore.location", "/Users/temp/certs/user.p12");
           // kubectl -n <namespace> get secret <Kafka User Name> -o jsonpath='{.data.user\.password}' | base64 -d
           kafkaProps.put("ssl.keystore.password", "************");
           // 内网访问无需配置，配置用于启用 nodeport 上的 TLS
           kafkaProps.put("ssl.endpoint.identification.algorithm", "");
           // ProducerRecord(String topic, K key, V value)
           ProducerRecord<String, String> record = new ProducerRecord<>("my-cluster-topic", "Precision Products", "China");
   
           try (KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps)) {
               // 忽略返回值，无法判断消息是否发送成功。
               producer.send(record);
               // Thread.sleep(2000000); // 延迟 2000 秒
           } catch (Exception e) {
               // 生产者发送消息到 Kafka 前遇到错误。
               e.printStackTrace();
           }
       }
   }

#3. 外网访问 - 有认证（SCRAM-SHA-512 协议），非加密传输

此时客户端与 Kafka 实例位于不同业务集群。Kafka 服务器开启认证，传输过程无需加密。客户端需获取 Kafka 实例的外网访问地址，以及 SCRAM-SHA-512 协议下的用户名和密码。

1. 可复制【Kafka 实例 - 访问方式】标签页中“集群外访问 - 明文传输”的连接地址，替换到实际代码片段中。

2. 可在【Kafka 实例 - 用户管理】标签页使用已有用户或创建新用户。

3. 客户端连接代码示例如下：

   package com.lanzhiwang.producer;
   import java.util.Properties;
   
   import org.apache.kafka.clients.producer.KafkaProducer;
   import org.apache.kafka.clients.producer.ProducerRecord;
   
   public class ExampleProducer {
       public static void main(String[] args) {
   
           Properties kafkaProps = new Properties();
           kafkaProps.put("bootstrap.servers", "192.168.176.13:31786,192.168.176.23:30535,192.168.176.27:30219");
           kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
           kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
   
           // 配置 SCRAM-SHA-512 属性
           kafkaProps.put("sasl.mechanism", "SCRAM-SHA-512");
           kafkaProps.put("security.protocol", "SASL_PLAINTEXT");
           kafkaProps.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"*************\" password=\"*************\";");
           // ProducerRecord(String topic, K key, V value)
           ProducerRecord<String, String> record = new ProducerRecord<>("my-cluster-topic", "Precision Products", "China");
   
           try (KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps)) {
               // 忽略返回值，无法判断消息是否发送成功。
               producer.send(record);
               // Thread.sleep(2000000); // 延迟 2000 秒
           } catch (Exception e) {
               // 生产者发送消息到 Kafka 前遇到错误。
               e.printStackTrace();
           }
       }
   }

#4. 外网访问 - 有认证（SCRAM-SHA-512 协议），加密传输

此时客户端与 Kafka 实例位于不同业务集群。Kafka 服务器开启认证，传输过程需加密。客户端需获取 Kafka 实例的外网访问地址、SCRAM-SHA-512 协议的用户名和密码，以及 TLS 协议传输证书。

1. 可复制【Kafka 实例 - 访问方式】标签页中“集群外访问 - 明文传输”的连接地址，替换到实际代码片段中。

2. 可在【Kafka 实例 - 用户管理】标签页使用已有用户或创建新用户。

3. TLS 协议传输及密码证书

   使用 kubectl 命令生成用户及密码证书，并将相关证书复制到客户端对应路径。具体命令如下：

   • 生成 CA 证书：

     $ kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12

   • 生成 CA 证书密码：

     $ kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d

4. 客户端连接代码示例如下：

   package com.lanzhiwang.producer;
   import java.util.Properties;
   
   import org.apache.kafka.clients.producer.KafkaProducer;
   import org.apache.kafka.clients.producer.ProducerRecord;
   
   public class ExampleProducer {
       public static void main(String[] args) {
   
           Properties kafkaProps = new Properties();
           kafkaProps.put("bootstrap.servers", "192.168.176.13:31786,192.168.176.23:30535,192.168.176.27:30219");
           kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
           kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
   
           kafkaProps.put("security.protocol", "SASL_SSL");
           kafkaProps.put("sasl.mechanism", "SCRAM-SHA-512");
           kafkaProps.put("ssl.truststore.location", "/Users/temp/certs/ca.p12");
           kafkaProps.put("ssl.truststore.password", "************");
           kafkaProps.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"*************\" password=\"*************\";");
           // 内网访问无需配置，配置用于启用 nodeport 上的 TLS
           kafkaProps.put("ssl.endpoint.identification.algorithm", "");
           // ProducerRecord(String topic, K key, V value)
           ProducerRecord<String, String> record = new ProducerRecord<>("my-cluster-topic", "Precision Products", "China");
   
           try (KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps)) {
               // 忽略返回值，无法判断消息是否发送成功。
               producer.send(record);
               // Thread.sleep(2000000); // 延迟 2000 秒
           } catch (Exception e) {
               // 生产者发送消息到 Kafka 前遇到错误。
               e.printStackTrace();
           }
       }
   }

#外网访问 - 有认证（TLS 协议），加密传输

此时客户端与 Kafka 实例位于不同业务集群。Kafka 服务器开启认证，传输过程需加密。客户端需获取 Kafka 实例的外网访问地址，以及 SCRAM-SHA-512 协议的用户名和密码，和 TLS 协议传输证书。

1. 可复制【Kafka 实例 - 访问方式】标签页中“集群外访问 - 明文传输”的连接地址，替换到实际代码片段中。

2. TLS 协议传输及密码证书

   使用 kubectl 命令生成用户及密码证书，并将相关证书复制到客户端对应路径。具体命令如下：

   • 生成 CA 证书：

     $ kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12

   • 生成 CA 证书密码：

     $ kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d

3. TLS 协议用户及密码证书

   使用 kubectl 命令生成传输及密码证书，并将相关证书复制到客户端对应路径。具体命令如下：

   • 生成用户证书：

     $ kubectl -n <namespace> get secret <Kafka User Name> -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12

   • 生成用户证书密码：

     $ kubectl -n <namespace> get secret <Kafka User Name> -o jsonpath='{.data.user\.password}' | base64 -d

4. 客户端连接代码示例如下：

   package com.lanzhiwang.producer;
   import java.util.Properties;
   
   import org.apache.kafka.clients.producer.KafkaProducer;
   import org.apache.kafka.clients.producer.ProducerRecord;
   
   public class ExampleProducer {
       public static void main(String[] args) {
   
           Properties kafkaProps = new Properties();
           kafkaProps.put("bootstrap.servers", "192.168.176.13:31786,192.168.176.23:30535,192.168.176.27:30219");
           kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
           kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
   
           // 配置 TLS/SSL 属性
           kafkaProps.put("security.protocol", "SSL");
           // kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.p12}' | base64 -d > ca.p12
           kafkaProps.put("ssl.truststore.location", "/Users/temp/certs/ca.p12");
           // kubectl -n <namespace> get secret <name>-cluster-ca-cert -o jsonpath='{.data.ca\.password}' | base64 -d
           kafkaProps.put("ssl.truststore.password", "************");
           // kubectl -n <namespace> get secret <Kafka User Name> -o jsonpath='{.data.user\.p12}' | base64 -d > user.p12
           kafkaProps.put("ssl.keystore.location", "/Users/temp/certs/user.p12");
           // kubectl -n <namespace> get secret <Kafka User Name> -o jsonpath='{.data.user\.password}' | base64 -d
           kafkaProps.put("ssl.keystore.password", "************");
           // 内网访问无需配置，配置用于启用 nodeport 上的 TLS
           kafkaProps.put("ssl.endpoint.identification.algorithm", "");
           // ProducerRecord(String topic, K key, V value)
           ProducerRecord<String, String> record = new ProducerRecord<>("my-cluster-topic", "Precision Products", "China");
   
           try (KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps)) {
               // 忽略返回值，无法判断消息是否发送成功。
               producer.send(record);
               // Thread.sleep(2000000); // 延迟 2000 秒
           } catch (Exception e) {
               // 生产者发送消息到 Kafka 前遇到错误。
               e.printStackTrace();
           }
       }
   }