Scala FetchRequestBuilder类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Scala›Scala教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Scala中kafka.api.FetchRequestBuilder类的典型用法代码示例。如果您正苦于以下问题：Scala FetchRequestBuilder类的具体用法？Scala FetchRequestBuilder怎么用？Scala FetchRequestBuilder使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

在下文中一共展示了FetchRequestBuilder类的2个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Scala代码示例。

示例1: KafkaConsumer

//设置package包名称以及导入依赖的类
package jp.gr.java_conf.massakai.kafka

import java.util.Collections

import kafka.api.{PartitionOffsetRequestInfo, FetchRequestBuilder}
import kafka.common.TopicAndPartition
import kafka.javaapi._
import kafka.javaapi.consumer.SimpleConsumer
import collection.JavaConversions._

object KafkaConsumer {
  def findLeader(bootstraps: Seq[Broker], topic: String, partition: Int): Option[PartitionMetadata] = {
    for (bootstrap <- bootstraps) {
      val consumer = new SimpleConsumer(bootstrap.host, bootstrap.port, 100000, 64 * 1024, "leaderLookup")
      val topics = Collections.singletonList(topic)
      val req = new TopicMetadataRequest(topics)
      val resp = consumer.send(req)
      val metadata: java.util.List[TopicMetadata] = resp.topicsMetadata
      for (topicMetadata: TopicMetadata <- metadata) {
        for (partitionMetadata: PartitionMetadata <- topicMetadata.partitionsMetadata) {
          if (partitionMetadata.partitionId == partition) {
            return Some(partitionMetadata)
          }
        }
      }
    }
    None
  }
}

case class KafkaConsumer(leadBroker: String, port: Int, soTimeout: Int, bufferSize: Int, clientName: String) {
  val consumer = new SimpleConsumer(leadBroker, port, soTimeout, bufferSize, clientName)


  def getMessages(topic: String, partition: Int, offset: Long, fetchSize: Int): FetchResponse = {
    val request = new FetchRequestBuilder()
      .clientId(clientName)
      .addFetch(topic, partition, offset, fetchSize)
      .build()
    consumer.fetch(request)
  }

  def getLastOffset(topic: String, partition: Int, whichTime: Long): Option[Long] = {
    val topicAndPartition = new TopicAndPartition(topic, partition)
    val requestInfo = new java.util.HashMap[TopicAndPartition, PartitionOffsetRequestInfo]()
    requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(whichTime, 1))
    val request = new OffsetRequest(requestInfo, kafka.api.OffsetRequest.CurrentVersion, clientName)
    val response = consumer.getOffsetsBefore(request)
    if (response.hasError) {
      println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition))
      None
    } else {
      Some(response.offsets(topic, partition)(0))
    }
  }

  // TODO: ?????
  // consumer.close()
}

开发者ID:massakai，项目名称:finagle-kafka-sample，代码行数:60，代码来源:KafkaConsumer.scala

示例2: ChunkConsumer

//设置package包名称以及导入依赖的类
package example.consumer

import kafka.api.FetchRequestBuilder
import kafka.consumer.SimpleConsumer

case class ChunkConsumer(topics: List[String], partition: Int = 0, offset: Long = 0L, fetchSize: Int = 100) extends Consumer(topics){

  private val clientId = kafkaConfig.getCustomString("consumer.clientId")

  val simpleConsumer = new SimpleConsumer(
    kafkaConfig.getCustomString("consumer.host"),
    kafkaConfig.getCustomInt("consumer.port"),
    kafkaConfig.getCustomInt("consumer.timeOut"),
    kafkaConfig.getCustomInt("consumer.bufferSize"),
    clientId)

  def read(): Iterable[String] = {

    //println(simpleConsumer.toString)
    val fetchRequest = new FetchRequestBuilder().clientId(clientId)
    for(topic <- topics) {
      fetchRequest.addFetch(topic, partition, offset, fetchSize)
    }

    val fetchResponse = simpleConsumer.fetch(fetchRequest.build())

    fetchResponse.data.values.flatMap { topic =>
      topic.messages.toList.map { mao =>
        val payload =  mao.message.payload

        //ugliest part of the code. Thanks to kafka
        val data = Array.fill[Byte](payload.limit)(0)
        payload.get(data)
        new String(data)
      }
    }
  }
}

开发者ID:alonsoir，项目名称:hello-kafka-twitter-scala，代码行数:39，代码来源:ChunkConsumer.scala

注：本文中的kafka.api.FetchRequestBuilder类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。