如何解决带有 spring kafka 模板工厂实现的 Kafka 生产者
我有一个简单的 rest api(方法 1),它使用 kafka-clients api 生成发送到 kafka 集群的消息。
Spring boot rest -> producer.send (kafka-clients lib) -> kafka 集群
spring boot rest->生产者工厂实现(单配置spring对象)->kafka模板发送(spring-kafka)->kafka集群
我观察到方法 2 比方法 1 花费更多的时间。例如,对于单个消息,方法 1 花费了 40 毫秒,方法 2 花费了将近 100 毫秒。
我想使用基于生产者工厂的实现来最小化推送消息所花费的时间。关于如何调整它的任何想法?
实现细节如下:(Producer factory)
@Configuration
public class KafkaConfig {
@Value("${bootstrap.servers}")
String bootStrapServers;
@Bean
public Map<String,Object> configs(){
Map<String,Object> properties = new HashMap<String,Object>();
properties.put("bootstrap.servers",bootStrapServers);
properties.put("acks","0");
properties.put("retries",0);
properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
return properties;
}
@Bean
public ProducerFactory<String,String> factory(){
return new DefaultKafkaProducerFactory<>(configs());
}
@Bean
public KafkaTemplate<String,String> template(){
return new KafkaTemplate<>(factory());
}
}
Controller :
@Autowired
private KafkaTemplate<String,String> template;
public ResponseEntity<String> producer(@PathVariable String topicName,@RequestBody String requestBody) throws JsonProcessingException {
try {
template.send(topicName,requestBody);
} catch (Exception ex) {
logger.error(ex);
} finally {
}
return ResponseEntity.ok().build();
}
解决方法
我确实看到的开销比我预期的要多(类似于您的结果)。我会做一些分析,看看它是否可以改进。
框架总是会增加一些开销,但最重要的是,与所有 Spring 项目一样,如果需要,您仍然可以下降到较低级别的 API。
@SpringBootApplication
public class So65791199Application {
public static void main(String[] args) {
SpringApplication.run(So65791199Application.class,args);
}
@Bean
public ApplicationRunner runner(KafkaTemplate<String,String> template,ProducerFactory<String,String> pf) {
return args -> {
StopWatch watch = new StopWatch();
ListenableFuture<SendResult<String,String>> future = template.send("so65791199","foo");
future.get(10,TimeUnit.SECONDS);
List<ListenableFuture<SendResult<String,String>>> futures = new LinkedList<>();
watch.start("template");
IntStream.range(0,10000).forEach(i -> {
futures.add(template.send("so65791199","foo"));
});
for (ListenableFuture<SendResult<String,String>> fut : futures) {
fut.get(10,TimeUnit.SECONDS);
}
watch.stop();
Producer<String,String> producer = new KafkaProducer<>(pf.getConfigurationProperties());
ProducerRecord<String,String> pr = new ProducerRecord<>("so65791199",null,"foo");
Future<RecordMetadata> fut = producer.send(pr);
fut.get(10,TimeUnit.SECONDS);
watch.start("raw producer");
List<Future<RecordMetadata>> futs = new LinkedList<>();
IntStream.range(0,10000).forEach(i -> {
futs.add(producer.send(new ProducerRecord<>("so65791199","foo")));
});
for (Future<RecordMetadata> futr : futs) {
futr.get(10,TimeUnit.SECONDS);
}
watch.stop();
producer.close();
System.out.println(watch.prettyPrint());
};
}
@Bean
public NewTopic topic() {
return TopicBuilder.name("so65791199").partitions(1).replicas(1).build();
}
}
StopWatch '': running time = 126595537 ns
---------------------------------------------
ns % Task name
---------------------------------------------
088742103 070% template
037853434 030% raw producer
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。