Apache Kafka：生產者不生產所有數據

Question

我是新的卡夫卡。我的要求是，我有兩個數據庫源和目標表。現在我想從源表中獲取數據並將其存儲到這些kafka之間的目標中，並將作爲生產者和消費者使用。我已經完成了代碼，但問題在於，當生產者生成數據時，有些數據會錯過生成。例如，如果我在源表中有100條記錄，那麼它不會生成全部100條記錄。我使用的卡夫卡0.10

MyProducer Config-

bootstrap.servers=192.168.1.XXX:9092,192.168.1.XXX:9093,192.168.1.XXX:9094 
acks=all 
retries=2 
batch.size=16384 
linger.ms=2 
buffer.memory=33554432 
key.serializer=org.apache.kafka.common.serialization.IntegerSerializer 
value.serializer=org.apache.kafka.common.serialization.StringSerializer 

我的製片代碼： -

public void run() { 
    SourceDAO sourceDAO = new SourceDAO(); 
    Source source; 
    int id; 
    try { 
     logger.debug("INSIDE RUN"); 
     List<Source> listOfEmployee = sourceDAO.getAllSource(); 
     Iterator<Source> sourceIterator = listOfEmployee.iterator(); 
     String sourceJson; 
     Gson gson = new Gson(); 
     while(sourceIterator.hasNext()) { 
      source = sourceIterator.next(); 
      sourceJson = gson.toJson(source); 
      id = source.getId(); 
      producerRecord = new ProducerRecord<Integer, String>(TOPIC, id, sourceJson); 
      producerRecords.add(producerRecord); 
     } 

     for(ProducerRecord<Integer, String> record : producerRecords) { 
      logger.debug("Producer Record: " + record.value()); 
      producer.send(record, new Callback() { 
       @Override 
       public void onCompletion(RecordMetadata metadata, Exception exception) { 
        logger.debug("Exception: " + exception); 
        if (exception != null) 
         throw new RuntimeException(exception.getMessage()); 
        logger.info("The offset of the record we just sent is: " + metadata.offset() 
          + " In Partition : " + metadata.partition()); 
       } 
      }); 
     } 
     producer.close(); 
     producer.flush(); 
     logger.info("Size of Record: " + producerRecords.size()); 
    } catch (SourceServiceException e) { 
     logger.error("Unable to Produce data...", e); 
     throw new RuntimeException("Unable to Produce data...", e); 
    } 
} 

我的消費配置： -

bootstrap.servers=192.168.1.XXX:9092,192.168.1.231:XXX,192.168.1.232:XXX 
group.id=consume 
client.id=C1 
enable.auto.commit=true 
auto.commit.interval.ms=1000 
max.partition.fetch.bytes=10485760 
session.timeout.ms=35000 
consumer.timeout.ms=35000 
auto.offset.reset=earliest 
message.max.bytes=10000000 
key.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer 

value.deserializer =組織。 apache.kafka.common.serialization.StringDeserializer

消費者代碼： -

public void doWork() { 
    logger.debug("Inside doWork of DestinationConsumer"); 
    DestinationDAO destinationDAO = new DestinationDAO(); 
    consumer.subscribe(Collections.singletonList(this.TOPIC)); 
    while(true) { 
     ConsumerRecords<String, String> consumerRecords = consumer.poll(1000); 
     int minBatchSize = 1; 
     for(ConsumerRecord<String, String> rec : consumerRecords) { 
      logger.debug("Consumer Recieved Record: " + rec); 
      consumerRecordsList.add(rec); 
     } 
     logger.debug("Record Size: " + consumerRecordsList.size()); 
     if(consumerRecordsList.size() >= minBatchSize) { 
      try { 
       destinationDAO.insertSourceDataIntoDestination(consumerRecordsList); 
      } catch (DestinationServiceException e) { 
       logger.error("Unable to update destination table"); 
      } 
     } 
    } 
} 

Answer 1

從什麼可以在這裏作用似乎我猜你沒有刷新或關閉生產。你應該注意到，發送異步運行，只是準備了一批是發以後（取決於生產者的配置）：

從kafka documentation

的send（）方法是異步的。當被調用時，它將記錄添加到暫掛記錄發送的緩衝區並立即返回。這使得生產者可以將單個記錄進行批處理以提高效率。

你應該嘗試是調用producer.close()您遍歷所有producerRecords後（順便說一句：你爲​​什麼要緩存，當你有很多的記錄，導致可能的問題，整個producerRecords）。

如果這樣做沒有幫助，應嘗試使用例如一個控制檯消費者找出缺少的東西。請提供更多代碼。生產者如何配置？你的消費者是怎樣的？什麼是producerRecords的類型？

希望有所幫助。