You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
70 lines
3.7 KiB
70 lines
3.7 KiB
package com.example.demokafka.util;
|
|
|
|
import org.apache.commons.lang.StringUtils;
|
|
import org.apache.kafka.common.serialization.StringDeserializer;
|
|
import org.springframework.beans.factory.annotation.Value;
|
|
import org.springframework.stereotype.Component;
|
|
|
|
import java.util.Properties;
|
|
|
|
/**
|
|
* @Description
|
|
* @Author ZhouWenTao
|
|
* @Date 2023/9/1 13:40
|
|
*/
|
|
@Component
|
|
public class KafkaUtil {
|
|
public static String trustestore="/opt/kafka/client.trustestore.p12";
|
|
|
|
@Value("${appCode}")
|
|
public String appCode;
|
|
@Value("${secret}")
|
|
public String secret;
|
|
public static String X_Consumer_Username = "dwVendor";
|
|
|
|
public static Properties getSafeProducerPro() {
|
|
Properties properties = new Properties();
|
|
properties.setProperty("bootstrap.servers", "10.0.10.153:29551,10.0.10.153:29552,10.0.10.153:29553");
|
|
properties.setProperty("acks", "all");
|
|
properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
|
|
properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
|
|
//properties.setProperty("sasl.jaas.config", String.format("org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";",appCode,secret));
|
|
//properties.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"e92224\" password=\"323236g6#\";");
|
|
properties.setProperty("sasl.mechanism", "SCRAM-SHA-512");
|
|
properties.setProperty("security.protocol", "SASL_SSL");
|
|
properties.setProperty("ssl.truststore.location", trustestore);
|
|
properties.setProperty("ssl.truststore.password", "pwd123");
|
|
properties.setProperty("buffer.memory", "33554432");
|
|
properties.setProperty("retries", "0");
|
|
properties.setProperty("ssl.endpoint.identification.algorithm", "");
|
|
return properties;
|
|
}
|
|
//keytool -keystore D:/keystore/client.trustestore.p12 -storepass fEVcb^QFB;IN$K5 -noprompt -alias client.trustestore -import -file D:/keystore/ca.crt -storetype PKCS12
|
|
|
|
//keytool -keystore D:/keystore/client.trustestore.p12 -storepass pwd123 -noprompt -alias client.trustestore -import -file D:/keystore/ca.crt -storetype PKCS12
|
|
public Properties getSafeConsumerPro() {
|
|
Properties properties = new Properties();
|
|
properties.setProperty("group.id", "isv-kafka");
|
|
properties.setProperty("key.deserializer", StringDeserializer.class.getName());
|
|
properties.setProperty("value.deserializer", StringDeserializer.class.getName());
|
|
if (StringUtils.isNotBlank(appCode)) {
|
|
System.out.println("使用鉴权");
|
|
properties.setProperty("bootstrap.servers", "10.0.10.153:29553,10.0.10.153:29554");
|
|
properties.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"dwVendor\" password=\"fEVcb^QFB;IN$K5\";");
|
|
properties.setProperty("sasl.mechanism", "SCRAM-SHA-512");
|
|
properties.setProperty("security.protocol", "SASL_SSL");
|
|
properties.setProperty("ssl.truststore.location", "/opt/kafka/client.trustestore.p12");
|
|
properties.setProperty("ssl.truststore.password", "pwd123");
|
|
}else{
|
|
properties.setProperty("bootstrap.servers", "10.0.10.153:29551,10.0.10.153:29552");
|
|
}
|
|
|
|
properties.setProperty("enable.auto.commit", "true");
|
|
properties.setProperty("auto.commit.interval.ms", "1000");
|
|
properties.setProperty("session.timeout.ms", "30000");
|
|
properties.setProperty("auto.offset.reset", "earliest");
|
|
properties.setProperty("ssl.endpoint.identification.algorithm", "");
|
|
return properties;
|
|
}
|
|
}
|