Apache Kafka：生产者不生产所有数据

问题描述：

我是新的卡夫卡。我的要求是，我有两个数据库源和目标表。现在我想从源表中获取数据并将其存储到这些kafka之间的目标中，并将作为生产者和消费者使用。我已经完成了代码，但问题在于，当生产者生成数据时，有些数据会错过生成。例如，如果我在源表中有100条记录，那么它不会生成全部100条记录。我使用的卡夫卡0.10Apache Kafka：生产者不生产所有数据

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"); 
      } 
     } 
    } 
}

答

从什么可以在这里作用似乎我猜你没有刷新或关闭生产。你应该注意到，发送异步运行，只是准备了一批是发以后（取决于生产者的配置）：

从kafka documentation

的send（）方法是异步的。当被调用时，它将记录添加到暂挂记录发送的缓冲区并立即返回。这使得生产者可以将单个记录进行批处理以提高效率。

你应该尝试是调用producer.close()您遍历所有producerRecords后（顺便说一句：你为什么要缓存，当你有很多的记录，导致可能的问题，整个producerRecords）。

如果这样做没有帮助，应尝试使用例如一个控制台消费者找出缺少的东西。请提供更多代码。生产者如何配置？你的消费者是怎样的？什么是producerRecords的类型？

希望有所帮助。

你能说什么消息丢失？第一个？最后一个？随机？ – TobiSH

Apache Kafka：生产者不生产所有数据

相关推荐