kafka-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Milind Parikh <milindpar...@gmail.com>
Subject Re: can high-level consumer api provide the method getting messages with non-block?
Date Fri, 02 Mar 2012 12:49:59 GMT
Using a timeout of >0 on the broker will throw a timeout on consumer.

/***********************
sent from my android...please pardon occasional typos as I respond @ the
speed of thought
************************/

On Mar 2, 2012 7:18 AM, <zlai_2001@sina.com> wrote:

hi:
    I use high-level consumer api to get message. But I want to use
non-block method. Below is the code:

      ConsumerConnector consumer =
 kafka.consumer.Consumer.createJavaConsumerConnector(consumerconfig);
   Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
   topicCountMap.put("topic", new Integer(1));
   Map<String, List<KafkaMessageStream<Message>>> consumerMap =
consumer.createMessageStreams(topicCountMap);
   KafkaMessageStream<Message> stream =  consumerMap.get("topic").get(0);
   ConsumerIterator<Message> it = stream.iterator();
   while(it.hasNext())
   {
    ByteBuffer buffer = it.next().payload();
    byte [] bytes = new byte[buffer.remaining()];
    buffer.get(bytes);
    System.out.println(new String(bytes));
   }

The problem is that when there are no message,the program is blocked at
"it.hasNext()";But I want to break out.  Are there some methods?
Thanks!

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message