有道笔记链接：点击这里

一.消费者(Consumer)测试demo

①安全协议(security.protocol)

②SASL mechanism配置(sasl.mechanism)

③授权信息(sasl.jaas.config)

public static void main(String[] args) {Properties properties = new Properties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.20.110:9092");properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");//安全协议properties.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");//SASL mechanism配置properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);properties.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"test\" password=\"test\";");//授权信息properties.put(ConsumerConfig.GROUP_ID_CONFIG, "testgroupId");//2. 创建消费者KafkaConsumer<Integer, String> consumer = new KafkaConsumer<Integer, String>(properties);//3. 订阅主题consumer.subscribe(Arrays.asList("topiccccc"));//4. 拉取主题内新增的数据while (true) {ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(5));// 拉取的超时时间for (ConsumerRecord<Integer, String> record : records) {// 1 HelloWorld  xxxx  0System.out.println(record.key() + " | " + record.value() + " | " + record.timestamp() + " | " + record.offset() + " | " + record.partition());}}}

二.生产者(Producer)测试demo

①安全协议(security.protocol)

②SASL mechanism配置(sasl.mechanism)

③授权信息(sasl.jaas.config)

 public static void main(String[] args) throws JSONException {SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");String date = format.format(new Date());//1. 创建配置对象 指定Producer的信息Properties properties = new Properties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.20.110:9092");properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");properties.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); // 对record的key进行序列化properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);properties.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"test\" password=\"test\";");//2. 创建Producer对象KafkaProducer<String, String> producer = new KafkaProducer<>(properties);//3. 发布消息ProducerRecord<String, String> record = new ProducerRecord<String, String>("topiccccc","","{\n" +"  \"data\": [\n" +"    {\n" +"      \"name\": null,\n" +"      \"age\": 22,\n" +"      \"sex\": \"ere\"\n" +"    }\n" +"  ]\n" +"}");producer.send(record);//4. 提交producer.flush();producer.close();}

Kafka SASL SCRAM授权java开发demo相关推荐

Kafka SASL SCRAM动态授权实现方案Java版
效果截图预览一.pom依赖  <dependency><groupId>org.apache.kafka</gro ...
Kafka SASL/SCRAM动态认证集群部署
Kafka SASL/SCRAM动态认证集群部署目的:配置SASL/PLAIN验证,实现了对Kafka的权限控制.但SASL/PLAIN验证有一个问题:只能在JAAS文件KafkaServer中配置 ...
Kafka SASL/SCRAM+ACL实现动态创建用户及权限控制
文章目录 SASL_SCRAM+ACL实现动态创建用户及权限控制使用SASL / SCRAM进行身份验证 1. 创建SCRAM Credentials 创建broker建通信用户(或称超级用户) 创 ...
kafka 简单安装以及java小demo
文章目录第1步,下载解压 kafka: 第2步,运行 kafka: 第3步,创建topic 第4步,生产者发送消息第5步,消费者接收消息使用 java 客户端 kafka 0.8.0版本demo ...
Kafka动态认证SASL/SCRAM配置+整合springboot配置
记录: zookeeper启动命令: [root@master-yzjgxh2571705819-1651919082731-99-0727183 bin]# ./zkServer.sh start ...
Kafka安装配置（SASL/SCRAM动态认证）
SASL/SCRAM验证方法可以在Kafka服务启动之后,动态的新增用户分并配权限,在业务变动频繁,开发人员多的情况下比SASL/PLAIN方法更加灵活. Zookeeper:3.4.13,kafka ...
Kafka动态认证SASL/SCRAM验证
一.目的配置SASL/PLAIN验证,实现了对Kafka的权限控制.但SASL/PLAIN验证有一个问题:只能在JAAS文件KafkaServer中配置用户,一但Kafka启动,无法动态新增用户.S ...
kafka集群搭建+权限认证（SASL/SCRAM）+整合springboot
本文介绍的的是kafka集群搭建.kafka权限认证(SASL/SCRAM).整合springboot项目. 1.创建kafka日志和zookeeper文件目录: /data/kafka/kafka- ...
Kafka集群搭建及SASL/SCRAM,ACL权限控制
Kafka集群搭建及SASL/SCRAM,ACL权限控制环境部署zookeeper集群及安全认证部署kafka集群及安全认证环境系统版本:Centos7.4 kafka版本:kafka_2. ...

Kafka SASL SCRAM授权java开发demo

有道笔记链接：点击这里

一.消费者(Consumer)测试demo

二.生产者(Producer)测试demo

Kafka SASL SCRAM授权java开发demo相关推荐

最新文章

热门文章