Scala ActorHelper类代码示例

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

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

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

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

示例1: ForwarderActor

//设置package包名称以及导入依赖的类
import akka.actor.{Actor, ActorSystem, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.Http.{IncomingConnection, ServerBinding}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.http.scaladsl.unmarshalling.PredefinedFromEntityUnmarshallers._
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model._
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import org.apache.spark.streaming.receiver.ActorHelper
import org.apache.spark._
import org.apache.spark.streaming._
import scala.concurrent.Future

class ForwarderActor extends Actor with ActorHelper {
  def receive = {
    case data: String => store(data)
  }
}

object Main extends App {
  implicit val system = ActorSystem("sparkDriverActorSystem")
  implicit val mat = ActorMaterializer()
  implicit val ec = system.dispatcher

  val host = "localhost"
  val restPort = 9090
  val actorName = "forwarder"
  val sparkDriverPort = 7777

  // Spark Streaming
  val conf = new SparkConf().setMaster("local[*]").setAppName("TestWebApp").set("spark.driver.port", sparkDriverPort.toString).set("spark.driver.host", host).set("spark.akka.heartbeat.interval", "1s")
  val ssc = new StreamingContext(conf, Seconds(30))
  ssc.actorStream[String](Props[ForwarderActor], actorName).print()

  // Akka HTTP
  val restSource: Source[IncomingConnection, Future[ServerBinding]] = Http().bind(interface = host, port = restPort)

  val handler: HttpRequest => HttpResponse = {
    case HttpRequest(GET, Uri.Path("/"), _, _, _) =>
      HttpResponse(entity = HttpEntity(ContentTypes.`text/html(UTF-8)`, "<html><body>Hi there!</body></html>"))

    case HttpRequest(POST, Uri.Path("/data"), _, entity, _) if entity.contentType == ContentTypes.`application/json` => {
      val url: String = s"akka.tcp://[email protected]$host:${sparkDriverPort + 1}/user/Supervisor0/$actorName"
      val s: Future[String] = Unmarshal(entity).to[String]
      s foreach (system.actorSelection(url) ! _)

      HttpResponse(200)
    }

    case _: HttpRequest =>
      HttpResponse(404, entity = "Page not found!")
  }

  // Start all the things
  ssc.start()

  val binding: Future[ServerBinding] = restSource.to(Sink.foreach { _ handleWithSyncHandler handler }).run()
}

开发者ID:NeQuissimus，项目名称:akka-http-spark-streaming，代码行数:60，代码来源:Main.scala

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