我对春靴和Kafka还不熟悉。我有一个简单的rest服务,调用该服务时会使用来自kafka主题的消息。有一个自定义配置类controller。想知道如何在运行时设置属性,例如MAX\u POLL\u RECORDS\u CONFIG、AUTO\u OFFSET\u RESET\u 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;
}
}
Kafka消费控制器。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 **********");
}
}
应用属性
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=消费者Factory.create消费者();
使用:
/**
* 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
参数覆盖使用者属性。
问题内容: 使用jQuery在运行时设置标签属性的最佳方法是什么? 另外,如何使用jQuery 获取标签属性的值? 问题答案: 要获取或设置HTML元素的属性,可以在jQuery中使用该函数。 要获取 href 属性,请使用以下代码: 要设置 href 属性,请使用以下代码: 在这两种情况下,请使用适当的选择器。如果已为anchor元素设置了类,请使用;如果已为anchor元素设置了id,请使用。
我想为Kafka服务器[9092端口]设置最大和最小内存值 假设最大值为2 GB,则内存使用量不应超过2GB,但当前超过它。 我有链接-https://kafka.apache.org/documentation/#java 从Apache站点配置 但是我不知道如何配置它。 我的目标是设置最大内存限制值,Kubernetes仪表板中的内存值不应超过最大内存限制值。 注意-设置最大内存限制值不应在K
我在本地有属性文件,我通过下面的方法在代码中读取字符串路径OfFile=System.getProperties("arg.get.prop"); 如何设置此系统属性以获取我的属性文件的自由路径server.xml
问题内容: 我正在Java中使用 Apache Commons-Net 库 我想要的是使用Java代码在客户端阶段设置FTP服务器的连接超时 例如: 如果我查看FTP服务器的vsftpd.conf设置文件, 有一个设置 我想知道FTP客户端是否可以使用Java代码控制此空闲超时 我尝试了以下方法,但并非所有方法都有效 请帮我 :) 问题答案: FTP客户端无法控制FTP服务器的设置。 但是您要问的
我有一套使用spring框架用Java写的服务。一些服务有预定的任务(使用spring的< code>@Scheduled注释)来做一些内务处理(生成处理过的数据,数据清理等。).由于这些计划任务,我不能运行服务的多个实例,因为所有实例都选择相同的任务,并且多次执行,导致重复/损坏的数据。为了解决这个问题,我想在任务执行时进行检查,并且只允许在一个实例上执行。我该怎么做?这个问题有更好的解决方法吗
我有一个Spring MVC,它使用一个外部库,我无法访问代码。此外部库使用标准system.get属性调用读取某些属性。我必须在使用服务之前设置这些值。 由于我的应用程序是SpringMVC应用程序,我不确定如何初始化这些属性。这是我到目前为止所做的,但由于某些原因,我认为这些值总是空的。 我将属性放在属性文件 然后,我在我的应用程序中添加了以下两行ontext.xml 我阅读了文档,为了设置初