Step By Step
服务开通
创建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
