我正在開發一個 SpringBoot 應用程式，它將 API 暴露給子/取消子 Kafka 主題。我們需要做的就是傳遞topic-nameAPI 呼叫，應用程式將訂閱它并消費訊息。
訂閱主題 API：
{FQDN}/sub/{topic-name}
退訂主題 API：
{FQDN}/unsub/{topic-name}
依賴
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.3.1.RELEASE</version>
</dependency>
我已經創建KafkaConsumerConfiguration了一些豆類（如下）。
@EnableKafka
@Configuration
public class KafkaConsumerConfiguration {
private static final String KAFKA_BOOTSTRAP_SERVER = "localhost:9092";
@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(kafkaConsumerFactory());
return factory;
}
@Bean
public ConsumerFactory<String, String> kafkaConsumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerProps());
}
@Bean
public Map<String, Object> consumerProps() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BOOTSTRAP_SERVER);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
return props;
}
}
我已經使用該ConcurrentMessageListenerContainer.start()方法來針對topic-id.
@Service
public class KafkaConsumerService {
// This map will be used to store the running consumers info.
// So that when we need to stop/unsub the topic, then we can get the container from this map
private static Map<String, ConcurrentMessageListenerContainer<String, String>> consumersMap = new HashMap<>();
@Autowired
private ConsumerFactory<String, String> kafkaConsumerFactory;
public void createConsumers(String topic,MessageListener messageListener) {
log.info("creating kafka consumer for topic {}", topic);
ContainerProperties containerProps = new ContainerProperties(topic);
containerProps.setPollTimeout(100);
ConcurrentMessageListenerContainer<String, String> container =new ConcurrentMessageListenerContainer<>(kafkaConsumerFactory, containerProps);
container.setupMessageListener(messageListener);
container.start();
consumersMap.put(topic, container);
log.info("created and started kafka consumer for topic {}", topic);
}
public void stopConsumer(String topic) {
log.info("stopping consumer for topic {}", topic);
ConcurrentMessageListenerContainer<String, String> container = consumersMap.get(topic);
if (container != null) {
container.stop();
log.info("consumer stopped!! Unsubscribed all topics or patterns and assigned partitions");
}
}
}
這個解決方案作業得很好。像，
- 這是使用 API 訂閱/取消訂閱 kafka 主題
- 也正確接收訊息。
但問題是，
每次我呼叫 API 訂閱主題時，它都會創建AdminClient.create(props)（下圖的第 336 行）
這會生成如下日志，
2022-04-26 01:31:11.676 INFO 26021 --- [nio-8888-exec-2] o.a.k.clients.admin.AdminClientConfig : AdminClientConfig values:
bootstrap.servers = [localhost:9092]
client.dns.lookup = default
client.id =
connections.max.idle.ms = 300000
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 120000
retries = 5
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = https
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
2022-04-26 01:31:12.043 INFO 26021 --- [nio-8888-exec-2] o.a.kafka.common.utils.AppInfoParser : Kafka version: 2.3.0
2022-04-26 01:31:12.043 INFO 26021 --- [nio-8888-exec-2] o.a.kafka.common.utils.AppInfoParser : Kafka commitId: fc1aaa116b661c8a
2022-04-26 01:31:12.043 INFO 26021 --- [nio-8888-exec-2] o.a.kafka.common.utils.AppInfoParser : Kafka startTimeMs: 1650918672031
我不希望 AdminClient 每次都創建這個實體。因為這樣，創建對該主題的訂閱大約需要 2 秒。這在我的用例中是不可接受的。
所需解決方案：
kafka 連接只會創建一次。然后我可以使用該連接 sub/unub 卡夫卡主題。這樣每次它都不會創建此連接，并且時間效率也有所提高。
謝謝。
uj5u.com熱心網友回復：
2.3.x 已經不支持很久了
https://spring.io/projects/spring-kafka#support
最后一個 2.3.x 版本是去年 7 月的 2.3.14。
admin 用于檢查主題是否存在；使用舊版本，由;中的missingTopicsFatal屬性控制 ContainerProperties在那個版本中確實如此。
對于現代版本（自 2.3.4 起），它是錯誤的，因此在您啟動容器時不會創建 AdminClient。
但是您確實需要升級到受支持的版本（建議 2.8.5 - 2.7.x 很快就會停止對 OSS 的支持）。
不支持重用具有不同（主題）的消費者；每次啟動容器時都會創建一個新的消費者。
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/465036.html