使用的是storm框架,并且是一个单例的kafka生产者;每次遍历一个1000元素的list,这样写是否会影响写入性能。
for (String value : list) {
kafkaProducer.send(new ProducerRecord<>(topic, value), new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
logger.error(exception);
}
}
});
}
kafkaProducer.flush();