卡夫卡的高级消费者API生命周期0.8

Question

我无法找到高级消费者的生命周期描述。我在0.8.2.2上，我不能使用来自kafka客户的“现代”消费者。这里是我的代码：

def consume(numberOfEvents: Int, await: Duration = 100.millis): List[MessageEnvelope] = { 
    val consumerProperties = new Properties() 
    consumerProperties.put("zookeeper.connect", kafkaConfig.zooKeeperConnectString) 
    consumerProperties.put("group.id", consumerGroup) 
    consumerProperties.put("auto.offset.reset", "smallest") 

    val consumer = Consumer.create(new ConsumerConfig(consumerProperties)) 

    try { 
     val messageStreams = consumer.createMessageStreams(
     Predef.Map(kafkaConfig.topic -> 1), 
     new DefaultDecoder, 
     new MessageEnvelopeDecoder) 

     val receiveMessageFuture = Future[List[MessageEnvelope]] { 
     messageStreams(kafkaConfig.topic) 
      .flatMap(stream => stream.take(numberOfEvents).map(_.message())) 
     } 

     Await.result(receiveMessageFuture, await) 
    } finally { 
     consumer.shutdown() 
    } 

我不清楚。我应该在每次检索消息后关闭消费者，还是保留实例并将其重新用于消息提取？我认为重用实例是正确的方式，但找不到一些文章/最佳实践。

我想重用消费者和/或messageStreams。它不适合我，我找不到原因。

如果我尝试重用messageStreams，我得到异常：

2017-04-17_19:57:57.088 ERROR MessageEnvelopeConsumer - Error while awaiting for messages java.lang.IllegalStateException: Iterator is in failed state 
java.lang.IllegalStateException: Iterator is in failed state 
    at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:54) 
    at scala.collection.IterableLike$class.take(IterableLike.scala:134) 
    at kafka.consumer.KafkaStream.take(KafkaStream.scala:25) 

发生在这里：

def consume(numberOfEvents: Int, await: Duration = 100.millis): List[MessageEnvelope] = { 
    try { 
     val receiveMessageFuture = Future[List[MessageEnvelope]] { 
     messageStreams(kafkaConfig.topic) 
      .flatMap(stream => stream.take(numberOfEvents).map(_.message())) 
     } 
     Try(Await.result(receiveMessageFuture, await)) match { 
     case Success(result) => result 
     case Failure(_: TimeoutException) => List.empty 
     case Failure(e) => 
      // ===> never got any message from topic 
      logger.error(s"Error while awaiting for messages ${e.getClass.getName}: ${e.getMessage}", e) 
      List.empty 

     } 
    } catch { 
     case e: Exception => 
     logger.warn(s"Error while consuming messages", e) 
     List.empty 
    } 
    } 

我试图创建messageStreams每次：

没有运气...

2017-04-17_20:02:44.236 WARN MessageEnvelopeConsumer - Error while consuming messages 
kafka.common.MessageStreamsExistException: ZookeeperConsumerConnector can create message streams at most once 
    at kafka.consumer.ZookeeperConsumerConnector.createMessageStreams(ZookeeperConsumerConnector.scala:151) 
    at MessageEnvelopeConsumer.consume(MessageEnvelopeConsumer.scala:47) 

Ha ppens这里：

def consume(numberOfEvents: Int, await: Duration = 100.millis): List[MessageEnvelope] = { 
    try { 

     val messageStreams = consumer.createMessageStreams(
     Predef.Map(kafkaConfig.topic -> 1), 
     new DefaultDecoder, 
     new MessageEnvelopeDecoder) 

     val receiveMessageFuture = Future[List[MessageEnvelope]] { 
     messageStreams(kafkaConfig.topic) 
      .flatMap(stream => stream.take(numberOfEvents).map(_.message())) 
     } 
     Try(Await.result(receiveMessageFuture, await)) match { 
     case Success(result) => result 
     case Failure(_: TimeoutException) => List.empty 
     case Failure(e) => 
      logger.error(s"Error while awaiting for messages ${e.getClass.getName}: ${e.getMessage}", e) 
      List.empty 

     } 
    } catch { 
     case e: Exception => 
     // ===> now exception raised here 
     logger.warn(s"Error while consuming messages", e) 
     List.empty 
    } 
    } 

UPD

我使用的基于迭代器的方法。它看起来是这样的：

// consumerProperties.put("consumer.timeout.ms", "100")  

private lazy val consumer: ConsumerConnector = Consumer.create(new ConsumerConfig(consumerProperties)) 

    private lazy val messageStreams: Seq[KafkaStream[Array[Byte], MessageEnvelope]] = 
    consumer.createMessageStreamsByFilter(Whitelist(kafkaConfig.topic), 1, new DefaultDecoder, new MessageEnvelopeDecoder) 


    private lazy val iterator: ConsumerIterator[Array[Byte], MessageEnvelope] = { 
    val stream = messageStreams.head 
    stream.iterator() 
    } 

    def consume(): List[MessageEnvelope] = { 
    try { 
     if (iterator.hasNext) { 
     val fromKafka: MessageAndMetadata[Array[Byte], MessageEnvelope] = iterator.next 
     List(fromKafka.message()) 
     } else { 
     List.empty 
     } 

    } catch { 
     case _: ConsumerTimeoutException => 
     List.empty 

     case e: Exception => 
     logger.warn(s"Error while consuming messages", e) 
     List.empty 
    } 
    } 

现在我想弄清楚它是否会自动提交偏移ZK ...

Answer 1

我的答案是最新的问题更新。迭代器方法可以像预期的那样为我工作。

Answer 2

恒关机导致不必要的消费群体重新平衡从而影响性能提升不少。请参阅本文以获得最佳做法：https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example