代码之家  ›  专栏  ›  技术社区  ›  Ahmet Karakaya

Kafka生产者请求超时设置

  •  3
  • Ahmet Karakaya  · 技术社区  · 8 年前

    我想设置请求超时以便添加请求。超时。ms参数。

    我在这个配置中缺少什么?我也需要修改服务器设置吗?

     public void init() {
              LOGGER.info("initializing  KafkaProducer: Topic Name: {}", topic);
              System.out.println("initializing  KafkaProducer: Topic Name: {}");
              Properties properties = new Properties();
              properties.put("bootstrap.servers", brokerList);
              properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
              properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
              properties.put("acks", "1");
              properties.put("retries", "3");
              properties.put("linger.ms", 5);
              properties.put("block.on.buffer.full", false);
              properties.put("request.timeout.ms", "1000");
              //properties.put("metadata.fetch.timeout.ms", 1000);
    
    
    
              producer = new KafkaProducer<>(properties);
           }
    
    
            public void produce(String txnLogStr) {
              ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, txnLogStr);
              producer.send(record, new ProducerCallback());   
    
           }
    
           private class ProducerCallback implements Callback {
              @Override
              public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                  System.out.println("onCompletion recordMetadata:"+recordMetadata.offset()+", e:"+e);
                 if (e != null && recordMetadata != null) {
                    LOGGER.error("Kafka Queue problem. Topic: {}", topic, e);
                    e.printStackTrace();
                 }else {
                     System.out.println("No error");
                 }
              }
           }
    

    卡夫卡版本 :kafka_2.11-0.10.2.0

     <dependencies>
        <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka_2.11</artifactId>
                <version>0.10.2.0</version>         
    
            </dependency>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>0.10.2.0</version> 
    
            </dependency>
    
    
      </dependencies>
    
    1 回复  |  直到 8 年前
        1
  •  2
  •   Ahmet Karakaya    8 年前

    使用以下属性设置

    properties.put("metadata.fetch.timeout.ms", "1000");
    
    推荐文章