Kafka配置JAAS

推薦本地測試使用，生產環境盡量使用KAFKA監聽，或者將此類方法添加定時開關與生產者同步避免循環打印

public static void main(String[] args) {Properties props = new Properties();props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");props.put("sasl.jaas.config","org.apache.kafka.common.security.plain.PlainLoginModule required username=\"***\" password=\"***\";");props.put("bootstrap.servers", "***");props.put("enable.auto.commit", "true");props.put("auto.commit.interval.ms", "1000");props.put("group.id", "***");props.put("session.timeout.ms", "60000");props.put("max.poll.records", 1000);props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");props.put("security.protocol", "SASL_PLAINTEXT");props.put("sasl.mechanism","PLAIN");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Arrays.asList("***"));//以下根據業務場景進行邏輯處理while (true) {long startTime = System.currentTimeMillis();ConsumerRecords<String, String> records = consumer.poll(1000);System.out.println(System.currentTimeMillis() - startTime);System.out.println("recieve message number is"+ records.count());for (ConsumerRecord<String, String> record : records) {System.out.printf("success" + "offset =%d,key =%s,value =%s,partition = %d %n",record.offset(), record.key(), record.value(),record.partition());}}}

總結

以上是生活随笔為你收集整理的Kafka配置JAAS的全部內容，希望文章能夠幫你解決所遇到的問題。

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