阿里雲Kafka SASL認證Quick Start

Step By Step

服務開通

SASL用戶授權
接入點對比

創建SASL用戶及授權

JAVA SDK測試

說明：目前SASL可以同時支持9093和9094端口，本地測試使用：公網 + 9093端口。

• PLAIN機制Code Sample

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import java.util.Properties;
import java.util.concurrent.Future;

public class demoForPlain {

    public static void main(String[] args) {

        Properties props = new Properties();
        props.put("sasl.jaas.config","org.apache.kafka.common.security.plain.PlainLoginModule required\n" +
                "username=\"******\"\n" +
                "password=\"*******\";");

        //與sasl路徑類似，該文件也不能被打包到jar中
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "C:\\AlikafkaDemoTest1\\src\\main\\resources\\kafka.client.truststore.jks");
        //根證書store的密碼，保持不變
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");

        // 設置接入點，即控制檯的實例詳情頁顯示的“默認接入點”
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "*********:9093");

        //接入協議，目前支持使用SASL_SSL協議接入
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        //SASL鑑權方式，保持不變
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");

        // Kafka消息的序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

        // 請求的最長等待時間
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10 * 1000);
        // 構造Producer對象，注意，該對象是線程安全的，一般來說，一個進程內一個Producer對象即可；
        // 如果想提高性能，可以多構造幾個對象，但不要太多，最好不要超過5個
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        // 構造一個Kafka消息
        String topic = "new_Topic"; //消息所屬的Topic，請在控制檯申請之後，填寫在這裡
        System.out.println("Topic: " + topic);
        String value = "This is the message's value from " + topic; //消息的內容

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, "Precision Products", value);//Topic Key Value
        try{
            for (int i = 0; i < 10; i++) {
                Future future = producer.send(record);
                future.get();//不關心是否發送成功，則不需要這行。
                System.out.println("i: " + i);
            }
        } catch(Exception e) {
            e.printStackTrace();//連接錯誤、No Leader錯誤都可以通過重試解決；消息太大這類錯誤kafkaProducer不會進行任何重試，直接拋出異常
        }
    }
}

• SCRAM機制Code Sample

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
import java.util.Properties;
import java.util.concurrent.Future;

public class demoForScram {

    public static void main(String[] args) {

        Properties props = new Properties();
        // sasl.jaas.config 參數配置
        props.put("sasl.jaas.config","org.apache.kafka.common.security.scram.ScramLoginModule required\n" +
                "username=\"*********）\"\n" +
                "password=\"*********\";");

        //與sasl路徑類似，該文件也不能被打包到jar中
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, "C:\\AlikafkaDemoTest1\\src\\main\\resources\\kafka.client.truststore.jks");
        //根證書store的密碼，保持不變
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");

        // 設置接入點，即控制檯的實例詳情頁顯示的“默認接入點”
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "******:9093");

        //接入協議，目前支持使用SASL_SSL協議接入
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        //SASL鑑權方式，保持不變
        props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");

        // Kafka消息的序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

        // 請求的最長等待時間
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10 * 1000);
        // 構造Producer對象，注意，該對象是線程安全的，一般來說，一個進程內一個Producer對象即可；
        // 如果想提高性能，可以多構造幾個對象，但不要太多，最好不要超過5個
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
        // 構造一個Kafka消息
        String topic = "new_Topic"; //消息所屬的Topic，請在控制檯申請之後，填寫在這裡
        System.out.println("Topic: " + topic);
        String value = "This is the message's value from " + topic; //消息的內容

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, "Precision Products", value);//Topic Key Value
        try{
            for (int i = 0; i < 10; i++) {
                Future future = producer.send(record);
                future.get();//不關心是否發送成功，則不需要這行。
                System.out.println("i: " + i);
            }
        } catch(Exception e) {
            e.printStackTrace();//連接錯誤、No Leader錯誤都可以通過重試解決；消息太大這類錯誤kafkaProducer不會進行任何重試，直接拋出異常
        }
    }
}

• 測試結果：

Topic: new_Topic
i: 0
i: 1
i: 2
i: 3
i: 4
i: 5
i: 6
i: 7
i: 8
i: 9

參考鏈接

SDK 公網接入
Kafka消息發送的三種模式