Kafka生产者用SpringKafka模板工厂实现

ghhaqwfi 于 2021-07-24 发布在 Java

关注(0)|答案(1)|浏览(445)

我有一个简单的restapi（方法1），它使用kafka客户机api生成发送到kafka集群的消息。
spring boot rest->producer.send（kafka clients lib）->kafka集群
此外，我还有另一个实现（方法2）
spring boot rest->producer factory implementation（单个配置spring对象）->kafka template send（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 apache-kafka spring-kafka

来源：https://stackoverflow.com/questions/65791199/kafka-producer-with-spring-kafka-template-factory-implementation

1条答案

按热度按时间

tquggr8v1#

我确实看到了比我预期的更多的开销（与您的结果类似）。我会做一些分析，看看是否可以改进。
框架总是会增加一些开销，但底线是，与所有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", 0, 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", 0, null, "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

赞(0）回复(0）举报 2021-07-24

我来回答

Kafka生产者用SpringKafka模板工厂实现

1条答案

相关问题

热门标签

最新问答