效果截圖預覽

一.pom依賴

?

<!-- kafka client --> <dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.5.0</version> </dependency> <!-- 登錄服務器所需依賴 --> <dependency><groupId>ch.ethz.ganymed</groupId><artifactId>ganymed-ssh2</artifactId><version>262</version> </dependency>

二.動態添加用戶

這里采用的是通過登錄服務器執行命令實現添加用戶，不是最優解決方案。

public static String execCommand(KafkaBroker broker,List<KafkaServerConfig> configs,KafkaUser user) {KafkaServerConfig config = configs.get(0);String host = config.getIp();int port = Integer.valueOf(config.getPort());String username = config.getUsername();String password = new EncryptUtil().AESdecode(config.getPassword(), EncryptUtil.AES_PRIVATE_KEY);String zookeepHost=broker.getZookeeper_ip()+":"+broker.getZookeeper_port();// 創建連接Connection conn = new Connection(host, port);// 啟動連接try {conn.connect();// 驗證用戶密碼conn.authenticateWithPassword(username, password);Session session = null;String userId = String.valueOf(user.getId());if (userId.equals("0")) {session = conn.openSession();String createUserCommand="./kafka-configs.sh "+ "--zookeeper "+zookeepHost+" --alter "+ "--add-config "+ "'SCRAM-SHA-256=[iterations=8192,password="+user.getPassword()+"],"+ "SCRAM-SHA-512=[password="+user.getPassword()+"]' "+ "--entity-type users --entity-name "+user.getUsername();session.execCommand("cd "+config.getKafka_install_path()+"/bin;source /etc/profile;"+createUserCommand);// 消費所有輸入流String inStr = printlnInfo(session.getStdout());String errStr = printlnInfo(session.getStderr());session.close();conn.close();return inStr;}else {session = conn.openSession();String deleteUserCommand_256="./kafka-configs.sh "+ "--zookeeper "+zookeepHost+" --alter "+ "--delete-config 'SCRAM-SHA-256' "+ "--entity-type users --entity-name "+user.getUsername();String deleteUserCommand_512="./kafka-configs.sh "+ "--zookeeper "+zookeepHost+" --alter "+ "--delete-config 'SCRAM-SHA-512' "+ "--entity-type users --entity-name "+user.getUsername();session.execCommand("cd "+config.getKafka_install_path()+"/bin;source /etc/profile;"+deleteUserCommand_512);// 消費所有輸入流String inStr = printlnInfo(session.getStdout());String errStr = printlnInfo(session.getStderr());session.close();session = conn.openSession();session.execCommand("cd "+config.getKafka_install_path()+"/bin;source /etc/profile;"+deleteUserCommand_256);inStr = printlnInfo(session.getStdout());errStr = printlnInfo(session.getStderr());session.close();conn.close();return inStr;}} catch (IOException e) { e.printStackTrace();return "連接服務器失敗，請檢查服務器配置是否正確";} }

三.動態為用戶授權

public static void main(String[] args) throws InterruptedException, ExecutionException {/*** 1.配置超管賬戶 ，這里的超管是在服務器配置kafka sasl scram時配置文件里配置的用戶 ，擁有所有權限*/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("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule "+ "required username=\"admin\" password=\"admin\";");AdminClient client = KafkaAdminClient.create(properties);/*** 2.添加topic acl信息，給shimengyuan賦予GtOPIC的讀權限，group_id為X-test-group*/List<AclBinding> aclBindings1 = Arrays.asList(new AclBinding(new ResourcePattern(ResourceType.TOPIC, "GtOPIC", PatternType.LITERAL),new AccessControlEntry("User:shimengyuan", "*", AclOperation.READ, AclPermissionType.ALLOW)));CreateAclsResult aclsResult = client.createAcls(aclBindings1);System.out.println(aclsResult.all().get());//若配置寫權限不用以下授權List<AclBinding> aclBindings = Arrays.asList(new AclBinding(new ResourcePattern(ResourceType.GROUP, "X-test-group", PatternType.LITERAL),new AccessControlEntry("User:shimengyuan", "*", AclOperation.READ, AclPermissionType.ALLOW)));CreateAclsResult aclsResult2 = client.createAcls(aclBindings);System.out.println(aclsResult2.all().get()); /*** 3.取消shimengyuan在GtOPIC的讀權限*/ AclBindingFilter bindingFilter = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, "X-test-group", PatternType.LITERAL),new AccessControlEntryFilter("User:shimengyuan", "*", AclOperation.READ, AclPermissionType.ALLOW));AclBindingFilter bindingFilter1 = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, "GtOPIC", PatternType.LITERAL),new AccessControlEntryFilter("User:shimengyuan", "*", AclOperation.READ, AclPermissionType.DENY));List<AclBindingFilter> del_aclBindings = new ArrayList<>();del_aclBindings.add(bindingFilter);del_aclBindings.add(bindingFilter1);client.deleteAcls(del_aclBindings, new DeleteAclsOptions());/*** 4.查看topic授權信息*/try {DescribeAclsResult describeAclsResult = client.describeAcls(new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, null, PatternType.LITERAL),new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.ANY)));DescribeAclsResult describeAclsResult1 = client.describeAcls(new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, null, PatternType.LITERAL),new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.ANY)));describeAclsResult.values().get().forEach(item -> {System.out.println("TOPIC授權信息:" + item);});describeAclsResult1.values().get().forEach(item -> {System.out.println("group授權信息:" + item);});} catch (InterruptedException | ExecutionException e) {e.printStackTrace();}}

?

?

?

總結

以上是生活随笔為你收集整理的Kafka SASL SCRAM动态授权实现方案Java版的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。