一、producer ACL实现
package com.hanwan.kafka.demo9;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.stream.IntStream;
/**
* @ClassName AclProducer
* @Description 模拟acl producer
* @Copyright: Copyright (c) 2018</p>
* @Company: www.lowan.com</ p>
* @Author hanwan
* @Date 2018/8/29 15:36
* @Version 1.0
**/
public class AclProducer {
private final static Logger LOGGER = LoggerFactory.getLogger(AclProducer.class);
static {
System.setProperty("java.security.auth.login.config", "c://workspace//idea//kafka-inaction//src//main//resources//producer_jaas.conf");
}
public static void main(String[] args) {
KafkaProducer<String, String> producer = new KafkaProducer<>(initProps());
IntStream.range(0, 10).forEach(i ->{
ProducerRecord<String, String> record = new ProducerRecord<>("acl-test", String.valueOf(i), "hello:"+i);
Future<RecordMetadata> future = producer.send(record);
try {
RecordMetadata recordMetadata = future.get();
LOGGER.info("THIS MESSAGE IS SEND DONE AND THE KEY IS {},OFFSET IS {}", i, recordMetadata.offset());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
});
producer.flush();
producer.close();
}
private static Properties initProps(){
final Properties prop = new Properties();
prop.put("bootstrap.servers","120.55.125.58:9093,120.26.198.248:9093,121.40.200.37:9093");
prop.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
prop.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
//producer-jaas.properties
prop.put("security.protocol", "SASL_PLAINTEXT");
prop.put("sasl.mechanism", "PLAIN");
return prop;
}
}
配置文件producer_jaas.conf
KafkaClient{
org.apache.kafka.common.security.plain.PlainLoginModule required
username="writer"
password="writer";
};
二、Consumer ACL代码实现
package com.hanwan.kafka.demo9;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.Properties;
/**
* @ClassName AclConsumer
* @Description 模拟acl consumer
* @Copyright: Copyright (c) 2018</p>
* @Company: www.lowan.com</ p>
* @Author hanwan
* @Date 2018/8/29 16:19
* @Version 1.0
**/
public class AclConsumer {
private final static Logger LOGGER = LoggerFactory.getLogger(AclConsumer.class);
static {
System.setProperty("java.security.auth.login.config", "C://workspace//idea//kafka-inaction//src//main//resources//consumer_jaas.conf");
}
public static void main(String[] args) {
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(loadProp());
consumer.subscribe(Collections.singleton("test1"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
records.forEach(record ->{
LOGGER.info("key {}, value {}, offset {}, partition {}", record.key(), record.value(), record.offset(), record.partition());
});
}
}
private static Properties loadProp(){
final Properties prop = new Properties();
prop.put("bootstrap.servers", "120.55.125.58:9093,120.26.198.248:9093,121.40.200.37:9093");
prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//consumer-jaas.properties
prop.put("security.protocol", "SASL_PLAINTEXT");
prop.put("sasl.mechanism", "PLAIN");
prop.put("group.id", "hw-group");
prop.put("client.id", "hw2");
prop.put("auto.offset.reset", "earliest");
return prop;
}
}
配置文件consumer_jaas.conf
KafkaClient{
org.apache.kafka.common.security.plain.PlainLoginModule required
username="reader"
password="reader";
};