java?SASL_SSL

帳號密碼?方式訪問 kafka

Producer Java Sample java生產者:

Properties props = new Properties();

props.put("bootstrap.servers",

"*******:9092,*******:9092");

props.put("acks", "all");//

props.put("retries", 3);

props.put("batch.size", 106384);

props.put("linger.ms", 1);

props.put("buffer.memory", 33554432);

props.put("key.serializer",

"org.apache.kafka.common.serialization.StringSerializer");

props.put("value.serializer",

"org.apache.kafka.common.serialization.StringSerializer");

props.put("security.protocol", "SASL_SSL");

props.put("ssl.truststore.location",

"D:/client_truststore.jks");

props.put("ssl.truststore.password", "WSO2_sp440");

props.put("sasl.mechanism", "SCRAM-SHA-512");

props.put("sasl.jaas.config",

"org.apache.kafka.common.security.scram.ScramLoginModule required

username='kaf_crm' password='xxxxxxx';");

//注意passwod結尾的分號一定不要漏

props.put("ssl.endpoint.identification.algorithm", "");

long sys = System.currentTimeMillis();

String contractId=CRM_ContractID

String payload = "payload";

Producer producer = new KafkaProducer<>(props);

//Synchronized Mode, Producer will wait and block until Kafka

Server return response

try{

Future future =producer.send(new

ProducerRecord<>("CRM_Contract", contractId, payload));//

(topic, key, payload),the second parameter is the key

future.get();//。 If not care whether success or failure , no

need this code

producer.close();

} catch(Exception e) {

e.printStackTrace();// Connection, No Leader error can be

resolved by retry; but too large message error will not re-try and

throw exception immediately

}

//Asynchronized mode, Producer not wait for response,

Background process of Producer submit message to Kafka server by

Batch size. It need callback to handle whether message is sent to

Kafka Server. If error happen ,need to log the exception.

try{

producer.send(new ProducerRecord<>("CRM_Contract",

contractId, payload),new Callback() {

public void onCompletion(RecordMetadata metadata, Exception e)

{

if(e != null) {

e.printStackTrace();

} else {

System.out.println("The offset of the record we just sent is:

" + metadata.offset());}}});

}catch(Exception e) {

e.printStackTrace();

}

Consumer Java Sample java消費者:

Properties props = new Properties();

props.put("bootstrap.servers", "*******:9092");

props.put("group.id", "wso2_sp");

props.put("enable.auto.commit", "false");

props.put("auto.commit.interval.ms", "1000");

props.put("session.timeout.ms", "30000");

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_SSL");

props.put("sasl.mechanism", "SCRAM-SHA-512");

props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,

"G:\\client_truststore.jks");

props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,

"WSO2_sp440");

props.put("sasl.jaas.config",

"org.apache.kafka.common.security.scram.ScramLoginModule required

username='kaf_xxx' password='xxxxx';");//注意passwod結尾的分號一定不要漏

props.put("ssl.endpoint.identification.algorithm", "");

KafkaConsumer consumer = new

KafkaConsumer<>(props);

String topic = "file_poc";

consumer.subscribe(Arrays.asList(topic));

while (true) {

ConsumerRecords records = consumer.poll(100);

for (ConsumerRecord record : records) {

System.out.printf("partition= %d, offset = %d, key = %s, value

= %s\n", record.partition(), record.offset(), record.key(),

record.value());

}

consumer.commitSync();

}

總結

以上是生活随笔為你收集整理的sasl java_javaSASL_SSL帐号密码方式访问kafka的全部內容，希望文章能夠幫你解決所遇到的問題。

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