java - 如何在java中优化.kafka生产者？

Question

@Configuration
public class KafkaConfiguration {
    @Value("${kafka.boot.server}")
    private String kafkaServer;

   @Bean    
   public KafkaTemplate<String,String> kafkaTemplate(){
       return new KafkaTemplate<>(producerConfig());}

   @Bean
   public ProducerFactory<String,String> producerConfig() {
       Map<String,Object> config= new HashMap<>();
       config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
       config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class );
       config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,JsonSerializer.class);          return new DefaultKafkaProducerFactory<>(config);
   }
}

使用kafka需要什么先决条件？您对发布消息有何建议？还有哪些其他可能的方法？

Answer 1

1. delivery.timeout.ms:如果你的情况是短时间内出现大量事件，这个值应该更高，因为当网络繁忙时，你的客户端会提示NetworkException，增加它你可以看到更少的NetworkException。

了解什么是delivery.timeout.ms:

https://cwiki.apache.org/confluence/display/KAFKA/KIP-91+Provide+Intuitive+User+Timeouts+in+The+Producer?source=post_page-----fa3910d9aa54----------------------#KIP-91ProvideIntuitiveUserTimeoutsinTheProducer-TestPlan

acks:如果您不需要丢失数据。您必须将其设置为全部。默认值为 1 并且领导者会将记录写入其本地日志，但会在不等待所有追随者完全确认的情况下做出响应。在这种情况下，如果领导者在确认记录后但在追随者复制记录之前立即失败，那么记录将丢失。

重试:取决于您的kafka客户端版本。现在默认的重试次数是Integer.Max，但对于早期版本，您需要将重试次数设置为更高的值，这样您的生产者就不会因为无法访问领导者分区这一简单异常而停止。

Exactly-once:如果您的应用需要exactly-once，则必须引用enable.idempotence和transactional.id

<

请注意，这里提到的配置应该可以在您的 java 客户端中找到相应的枚举

生产者设置的进一步引用: https://docs.confluent.io/current/installation/configuration/producer-configs.html