package com.example.zxweb.utils; import org.apache.commons.lang3.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 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"); 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", trustestore); properties.setProperty("ssl.truststore.password", "pwd123"); }else{ properties.setProperty("bootstrap.servers", "10.0.10.153:29551,10.0.10.153:29552"); } //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("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(boolean sslFlag) { 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 (sslFlag) { 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", trustestore); 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", "9000");//一次session的连接超时时间 properties.setProperty("heartbeat.interval.ms", "3000");// 心跳时间,一般为超时时间的1/3,Consumer在被判定为死亡之前,能够发送至少 3 轮的心跳请求 properties.setProperty("max.poll.interval.ms", "9000");//每隔多长时间去拉取消息。合理设置预期值,尽量但间隔时间消费者处理完业务逻辑,否则就会被coordinator判定为死亡,踢出Consumer Group,进行Rebalance properties.setProperty("max.poll.records", "100");//一次从拉取出来的数据条数。根据消费业务处理耗费时长合理设置,如果每次max.poll.interval.ms 设置的时间较短,可以max.poll.records设置小点儿,少拉取些,这样不会超时。 properties.setProperty("auto.offset.reset", "earliest"); properties.setProperty("ssl.endpoint.identification.algorithm", ""); return properties; } }