如何解决Spring Boot REST服务在运行时在控制器内部设置Kafka主题属性
我是Spring Boot和kafka的新手。我有一个简单的休息服务，当被调用时会消耗来自kafka主题的消息。有一个自定义配置类，控制器。想知道如何在运行时设置属性，例如 MAX_POLL_RECORDS_CONFIG ， AUTO_OFFSET_RESET_CONFIG 等。我看到KafkaConfig仅在启动时执行。如何在运行时设置上述属性，即基于某些查询参数要设置这些属性，有关如何在运行时在 getMessagesFromKafkaTopic 方法中进行设置的任何想法。这是我的控制器kafkaConfig
KafkaConfig.java
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
@Configuration
@EnableKafka
public class KafkaConfig {
@Value("${kafka.con.server}")
private String server;
@Value("${kafka.con.groupid}")
private String gid;
@Value("${kafka.con.enablecommit}")
private String enablecommit;
@Value("${kafka.con.ommitinterval}")
private String commitint;
@Value("${kafka.con.sessiontimeout}")
private String timeout;
@Value("${kafka.con.maxrecordspoll}")
private String maxPollRecords;
@Value("${kafka.con.offsetReset}")
private String offsetReset;
/**
* ConsumerFactory
* @return
*/
@Bean
public ConsumerFactory<Object,Object> consumerFactory(){
Map<String,Object> configs = new HashMap<String,Object>();
configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,server);
configs.put(ConsumerConfig.GROUP_ID_CONFIG,gid);
configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,enablecommit);
configs.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,commitint);
configs.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,timeout);
configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,maxPollRecords);
configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,offsetReset);
configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);
return new DefaultKafkaConsumerFactory<Object,Object>(configs);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String,Object> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String,Object> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
}
KafkaConsumerController.java
import java.text.Format;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Collections;
import java.util.Date;
import java.util.Properties;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping(value = "/kafka")
public class KafkaConsumerController {
@Autowired
private ConsumerFactory<Object,Object> consumerFactory;
@Autowired
private ConcurrentKafkaListenerContainerFactory concurrentKafkaListenerContainerFactory;
@RequestMapping(value = "/consume",method = RequestMethod.GET)
public void getMessagesFromKafkaTopic(@RequestParam("batchsize") int batchsize) {
System.out.println("********* batchsize " + batchsize);
//here would like to set MAX_POLL_RECORDS_CONFIG to the value coming in as param
Consumer<Object,Object> con = null;
try {
con = consumerFactory.createConsumer();
con.subscribe(Collections.singleton("demotopic"));
int count = 0;
while (true) {
ConsumerRecords<Object,Object> records = con.poll(Duration.ofSeconds(3));
System.out.println("****** Record Count ******* : " + records.count());
if(records.count() == 0) {
count++;
if(count > 3) {
break;
}
else
continue;
}
for (ConsumerRecord<Object,Object> record : records) {
System.out.println("Message: " + record.value());
System.out.println("Message offset: " + record.offset());
System.out.println("Message headers: " + record.timestamp());
Date date = new Date(record.timestamp());
Format format = new SimpleDateFormat("yyyy MM dd HH:mm:ss.SSS");
System.out.println("Message date: " + format.format(date));
}
con.commitSync();
}
} catch (Exception e) {
e.printStackTrace();
}finally {
con.close();
}
System.out.println("********* END **********");
}
}
application.properties
kafka.con.server=localhost:9092
kafka.con.groupid=my-first-consumer-group
kafka.con.enablecommit=false
kafka.con.ommitinterval=1000
kafka.con.sessiontimeout=30000
kafka.con.maxrecordspoll=5
kafka.con.offsetReset=earliest
解决方法
代替con = consumerFactory.createConsumer();使用：
/**
* Create a consumer with an explicit group id; in addition,the
* client id suffix is appended to the clientIdPrefix which overrides the
* {@code client.id} property,if present. In addition,consumer properties can
* be overridden if the factory implementation supports it.
* @param groupId the group id.
* @param clientIdPrefix the prefix.
* @param clientIdSuffix the suffix.
* @param properties the properties to override.
* @return the consumer.
* @since 2.2.4
*/
Consumer<K,V> createConsumer(@Nullable String groupId,@Nullable String clientIdPrefix,@Nullable String clientIdSuffix,@Nullable Properties properties);
这允许您通过properties参数覆盖使用者属性。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。