我有一个java使用者,它不断地运行着大量的数据和消息。。有时,程序会被卡在消费者身上。。它看起来像死锁或无限循环,但我不确定。没有抛出异常,程序似乎工作正常,尽管它不轮询实际数据(这个问题只发生在java程序中。。我运行console consumer,效果很好)
唯一的办法就是重新启动程序。。。
public class Main {
Properties consumerComfig = new Properties ();
consumerConfig.put(ConsumerConfig. KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"Org.apache.kafka.comnect.json.JsonDeserializer");
consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "some server")
consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID());
consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"):
consumerConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
consumerConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,10000)
consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest"):
consumer = new KafkaConsumer<String, JsonNode> (consumerConfig);
public void consumeAndSendToCorrelator () {
logger.info ("Starting kafka consumer, listening to topics stringBuilder.toString ()");
try {
consumer.subscribe("INPUT TOPICS");
while (true) {
ConsumerRecords<String, JsonNode> records = consumer.poll(8000);
logger.debug("Trying to pull records from consumer");
JsonNode value = null;
for (Consumer Record<String, JsonNode > irecord :records){
try {
value = record.value();
if (value == null)
continue;
//Somme thread that process the message
correlatorActor.tell(value, ActorRef.noSender);
} catch (JsonProcessingException e) {
logger.error("Could not parse incoming json" + value.toString());
} catch (IOException e) {
logger.error("IOException occurred", e);
} catch (Exception e) {
logger.error("Error occurred", e);
}
consumer.commitSync();
}
}
}catch(WakeupException e) {
logger.info("closed consumer");
}finally {
consumer.close()
}
}
}
我用的是Kafka0.10你知道会发生什么吗?我能做些什么来修复它?
暂无答案!
目前还没有任何答案,快来回答吧!