Today, I change the partition number from 2 to 6 in order to improve the
consume performance, but when I restart the consumer, it can sometime
consume some or nothing. Actually, producers are producing data all the
time.
I change back the partition, not work, rmr zookeeper consumer info, still
not work.Can only consume some data and then stop, I don't know why,
anyone can help ? The following is my storm code, Thanks a lot.
public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
// TODO Auto-generated method stub
_collector = collector;
consumerConnector =
Consumer.createJavaConsumerConnector(KafkaProperties.createConsumerConfig(groupId));
Map<String, List<KafkaMessageStream<Message>>> topicMessageStreams =
consumerConnector.createMessageStreams(ImmutableMap.of(topic,
numStreams));
streams = topicMessageStreams.get(topic);
executor = Executors.newFixedThreadPool(4);
}
@Override
public void nextTuple() {
// TODO Auto-generated method stub
boolean hasMessage = true;
if(streams != null && streams.size() > 0){
int i = 0;
for(final KafkaMessageStream<Message> stream : streams){
ConsumerIterator<Message> it = stream.iterator();
int count = 0;
i++;
//stream是blocking queue ? while(it.hasNext())会死循环, log一直看不到
try{
while(it != null && it.hasNext()){
String msg = getMessage(it.next());
msg = Debug.appendTimestamp(msg);
_collector.emit(new Values(msg), new KafkaMessageId(msg, 1));
Debug.incr(topic + "_" + MultiKafkaSpout.class.getSimpleName(), 1);
Debug.log(this.getClass().getSimpleName(), msg);
count++;
if(count >= LOOP_COUNT){
break;
}
}
}catch(ConsumerTimeoutException ex){
LOG.warn(topic + " : no message to consume");
hasMessage = false;
}catch(Exception ex){
//设置了consumer timeout，避免blocking queue
LOG.info(topic + " : " + ex.getMessage());
}
LOG.info(String.format("stream#%d emits : %d messages", i, count));
}
}else{
LOG.warn("no streams available, retry to get streams");
Map<String, List<KafkaMessageStream<Message>>> topicMessageStreams =
consumerConnector.createMessageStreams(ImmutableMap.of(topic,
numStreams));
streams = topicMessageStreams.get(topic);
}
if(! hasMessage){
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
LOG.warn("Thread sleep Exception", e);
}
}
}
--
have a good day!
chenshang'an
