• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    公众号

Scala Consumer类代码示例

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

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



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

示例1: KafkaClient

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

import java.util.Properties
import java.util.concurrent.Executors

import akka.actor.{Props, DeadLetter, ActorSystem}
import kafka.consumer.{Consumer, ConsumerConfig}
import scala.concurrent.{ExecutionContext, Future}

object KafkaClient {

  val config = {
    val properties = new Properties()
    properties.put("bootstrap.servers", "localhost:9092")
    properties.put("group.id", "pogo_consumer")
    properties.put("auto.offset.reset", "largest")
    properties.put("zookeeper.connect", "localhost:2181")
    properties.put("zookeeper.session.timeout.ms", "400")
    properties.put("zookeeper.sync.time.ms", "200")
    properties.put("auto.commit.interval.ms", "500")
    new ConsumerConfig(properties)
  }

  // Our actor system managing our actors
  val system = ActorSystem("es-sharpshooter")

  // Taking care of dead letters
  system.eventStream.subscribe(system.actorOf(Props[IndexService], "dead-letters"), classOf[DeadLetter])

  // Dedicated Kafka Execution context
  implicit val KafkaContext = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(20))

  def start = {
    // Starting our consumer
    val consumer = Consumer.create(config)

    val topics = Map(
      "pokemons" -> 1,
      "spawnpoints" -> 1
    )

    val streams = consumer.createMessageStreams(topics)

    // Start the consumer asynchronously
    Future {
      streams.get("pokemons").get.foreach(PokemonService.cycle(system))
    } onFailure { case ec => println(ec) }
    Future {
      streams.get("spawnpoints").get.foreach(SpawnService.cycle(system))
    } onFailure { case ec => println(ec) }
  }
} 
开发者ID:fiahil,项目名称:Talks,代码行数:53,代码来源:KafkaClient.scala


示例2: KafkaConsumer

//设置package包名称以及导入依赖的类
package Services



import java.util.Properties

import kafka.consumer.{Consumer, ConsumerConfig, ConsumerTimeoutException, Whitelist}
import kafka.serializer.DefaultDecoder


class KafkaConsumer(topic: String, groupId: String, zookeeperConnect: String) {


  private val props = new Properties()

  props.put("group.id", groupId)
  props.put("zookeeper.connect", zookeeperConnect)
  props.put("auto.offset.reset", "smallest")
  props.put("consumer.timeout.ms", "500")
  props.put("auto.commit.interval.ms", "500")

  private val config = new ConsumerConfig(props)
  private val connector = Consumer.create(config)
  private val filterSpec = new Whitelist(topic)
  private val streams = connector.createMessageStreamsByFilter(filterSpec, 1, new DefaultDecoder(), new DefaultDecoder())(0)

  lazy val iterator = streams.iterator()

  def read() =
    try {
      if (hasNext) {
        println("Getting message from queue.............")
        val message = iterator.next().message()
        Some(new String(message))
      } else {
        None
      }
    } catch {
      case ex: Throwable =>
        ex.printStackTrace()
        None
    }

  private def hasNext(): Boolean =
    try
      iterator.hasNext()
    catch {
      case timeOutEx: ConsumerTimeoutException =>
        false
      case ex: Throwable =>
        println("Getting error when reading message ")
        false
    }

  def close(): Unit = connector.shutdown()

} 
开发者ID:pranjut,项目名称:consul-kafka-microservices,代码行数:58,代码来源:KafkaConsumer.scala


示例3: KafkaConsumer

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

import java.util.Properties

import kafka.consumer.{Consumer, ConsumerConfig, ConsumerTimeoutException, Whitelist}
import kafka.serializer.DefaultDecoder


class KafkaConsumer(topic: String, groupId: String, zookeeperConnect: String) {



  private val props = new Properties()

  props.put("group.id", groupId)
  props.put("zookeeper.connect", zookeeperConnect)
  props.put("auto.offset.reset", "smallest")
  //2 minute consumer timeout
  props.put("consumer.timeout.ms", "120000")
  //commit after each 10 second
  props.put("auto.commit.interval.ms", "10000")
  private val config = new ConsumerConfig(props)
  private val connector = Consumer.create(config)
  private val filterSpec = new Whitelist(topic)
  private val streams = connector.createMessageStreamsByFilter(filterSpec, 1, new DefaultDecoder(), new DefaultDecoder())(0)

  lazy val iterator = streams.iterator()

  def read(): Option[String] =
    try {
      if (hasNext) {
        println("Getting message from queue.............")
        val message = iterator.next().message()
        Some(new String(message))
      } else {
        None
      }
    } catch {
      case ex: Exception =>
        ex.printStackTrace()
        None
    }

  private def hasNext(): Boolean =
    try
      iterator.hasNext()
    catch {
      case timeOutEx: ConsumerTimeoutException =>
        false
      case ex: Exception =>
        ex.printStackTrace()
        println("Getting error when reading message ")
        false
    }

  def close(): Unit = connector.shutdown()

} 
开发者ID:knoldus,项目名称:activator-kafka-scala-producer-consumer.g8,代码行数:59,代码来源:KafkaConsumer.scala


示例4: KafkaConsumer

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

import java.util.Properties

import kafka.consumer.{Consumer, ConsumerConfig, ConsumerTimeoutException, Whitelist}
import kafka.serializer.DefaultDecoder
import play.api.Logger


class KafkaConsumer(topic: String, groupId: String, zookeeperConnect: String) {



  private val props = new Properties()

  props.put("group.id", groupId)
  props.put("zookeeper.connect", zookeeperConnect)
  props.put("auto.offset.reset", "smallest")
  //2 minute consumer timeout
  props.put("consumer.timeout.ms", "120000")
  //commit after each 10 second
  props.put("auto.commit.interval.ms", "10000")
  private val config = new ConsumerConfig(props)
  private val connector = Consumer.create(config)
  private val filterSpec = new Whitelist(topic)
  private val streams = connector.createMessageStreamsByFilter(filterSpec, 1, new DefaultDecoder(), new DefaultDecoder())(0)

  lazy val iterator = streams.iterator()

  def read(): Option[String] =
    try {
      if (hasNext) {
        Logger.info("Getting message from queue.............")
        val message = iterator.next().message()
        Some(new String(message))
      } else {
        None
      }
    } catch {
      case ex: Exception =>
        ex.printStackTrace()
        None
    }

  private def hasNext(): Boolean =
    try
      iterator.hasNext()
    catch {
      case timeOutEx: ConsumerTimeoutException =>
        false
      case ex: Exception =>
        Logger.error("Getting error when reading message ",ex)
        false
    }

  def close(): Unit = connector.shutdown()

} 
开发者ID:satendrakumar,项目名称:play-kafka-example,代码行数:59,代码来源:KafkaConsumer.scala


示例5: KafkaConsumer

//设置package包名称以及导入依赖的类
package services

import java.util.Properties

import akka.actor.{Props, Actor}
import kafka.consumer.{KafkaStream, Consumer, ConsumerConfig}
import play.Logger

object KafkaConsumer {
  def props(topic: String): Props = Props(new KafkaConsumer(topic))
}

class KafkaConsumer(topic: String) extends Actor {
  val props = new Properties()
  props.put("auto.commit", "true")
  //props.put("zookeeper.connect", "localhost:2181")
  props.put("zookeeper.connect", "zookeeper1:2181")

  def receive = {
    case group: String => {
      props.put("group.id", group)

      val connector = Consumer.create(new ConsumerConfig(props))
      val stream: KafkaStream[Array[Byte], Array[Byte]] = connector.createMessageStreams(Map(topic -> 1)).get(topic).get(0)
      try {
        for (message <- stream) {
          try {
            Logger.debug(new String(message.message()))
          } catch {
            case e: Throwable => Logger.error("error processing message, skipping and resume consumption: " + e)
          }
        }
      }
    }
  }
} 
开发者ID:panchul,项目名称:sb_scala,代码行数:37,代码来源:KafkaConsumer.scala


示例6: KafkaConsumer

//设置package包名称以及导入依赖的类
package controllers

import java.util.Properties

import kafka.consumer.{Consumer, ConsumerConfig, ConsumerTimeoutException, Whitelist}
import kafka.serializer.DefaultDecoder


class KafkaConsumer(topic: String, groupId: String, zookeeperConnect: String) {



  private val props = new Properties()

  props.put("group.id", groupId)
  props.put("zookeeper.connect", zookeeperConnect)
  props.put("auto.offset.reset", "smallest")
  //2 minute consumer timeout
  props.put("consumer.timeout.ms", "120000")
  //commit after each 10 second
  props.put("auto.commit.interval.ms", "10000")
  private val config = new ConsumerConfig(props)
  private val connector = Consumer.create(config)
  private val filterSpec = new Whitelist(topic)
  private val streams = connector.createMessageStreamsByFilter(filterSpec, 1, new DefaultDecoder(), new DefaultDecoder())(0)

  lazy val iterator = streams.iterator()

  def read(): Option[String] =
    try {
      if (hasNext) {
        println("Getting message from queue.............")
        val message = iterator.next().message()
        Some(new String(message))
      } else {
        None
      }
    } catch {
      case ex: Exception =>
        ex.printStackTrace()
        None
    }

  private def hasNext(): Boolean =
    try
      iterator.hasNext()
    catch {
      case timeOutEx: ConsumerTimeoutException =>
        false
      case ex: Exception =>
        ex.printStackTrace()
        println("Getting error when reading message ")
        false
    }

  def close(): Unit = connector.shutdown()

} 
开发者ID:krishnangc,项目名称:kafka-scala-quartz,代码行数:59,代码来源:KafkaConsumer.scala



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


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Scala GraphiteReporter类代码示例发布时间:2022-05-23
下一篇:
Scala SparkListener类代码示例发布时间:2022-05-23
热门推荐
热门话题
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap