阿里云Kafka SASL认证Quick Start-阿里云开发者社区

开发者社区> 云服务技术课堂> 正文

阿里云Kafka SASL认证Quick Start

简介: 借助消息队列Kafka版的ACL,用户可以按需为SASL用户赋予不同访问消息队列Kafka版资源的权限,实现权限分割。阿里云专业版Kafka实例目前已经可以支持SASL认证方式,本文主要演示SASL的开通及使用。

Step By Step

服务开通

SASL用户授权
接入点对比

创建SASL用户及授权

图片.png

图片.png

JAVA SDK测试

说明:目前SASL可以同时支持9093和9094端口,本地测试使用:公网 + 9093端口。
图片.png

  • 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消息发送的三种模式

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:

云服务技术课堂,各类技术课程、最佳实践输出,来好好听课吧!

官方博客