by : digitalevers kafka, scala, 大数据

模拟点击-Akka定时向Kafka发送模拟数据（定时器的另类实现）

Spark的流式计算经常需要模拟流数据，这时就需要使用定时器每隔一秒或每隔几秒向Kafka写入模拟数据，模拟数据的方法及发送策略的制定可以参考这一篇，大体思想就是设置一天模拟的总数据量，然后使用一定的算法分解到每一秒中，再把分解好的数字序列存放到一个大的数组中，最后使用定时器从头到尾扫描这个数组，逐一取出再组装相应的模拟数据量就是了。

可以使用原生 Java 的 Timer 定时调度工具来实现这个定时器，但是这里采用一种另类的方式来实现——采用Akka的actor来实现。整体来说比较简单，直接上代码

package com.datacrafts.digitalevers.simulationClick

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.datacrafts.digitalevers.simulationClick.common._
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration
import scala.concurrent.duration.FiniteDuration

class sendClick extends Actor{

  override def receive: Receive = {
    case "start" =>{
      println("客户端启动")
      import context.dispatcher
      context.system.scheduler.scheduleWithFixedDelay(FiniteDuration(0,duration.MILLISECONDS),FiniteDuration(1000,duration.MILLISECONDS), self, sendData2Kafka)
    }
    case sendData2Kafka=>{
      println("客户端向Kafka发送数据~")
    }
  }

}


object sendClick {
  def main(args: Array[String]): Unit = {
    val host = "127.0.0.1"
    val port = 10006

    val config = ConfigFactory.parseString(s"""
                                              |akka.actor.provider="akka.remote.RemoteActorRefProvider"
                                              |akka.actor.allow-java-serialization = on
                                              |akka.remote.artery.canonical.hostname=$host
                                              |akka.remote.artery.canonical.port=$port""".stripMargin)

    val actorSystem = ActorSystem("work",config)
    val sparkWorkRef: ActorRef = actorSystem.actorOf(Props(new sendClick),"sparkWork")

    sparkWorkRef ! "start"
  }
}

这里采用的是 Akka2.6.8，其他版本的配置参数可能会有些许不同，读者可自行查阅文档。host 和 port 可配可不配，这里还是配置上了，如果不配置的话，Akka会默认当前主机，同时随机打开一个端口供 actor 使用。

如代码所见，这个例程去掉了 Akka 与服务端的交互，而 Akka 的一般应用场景就是与服务端的交互。这里只利用到 actor 对自身发送消息这一机制

当actor启动后，便启动定时器每隔一秒向自身发送一个 sendData2Kafka 对象消息，这个空消息对象定义如下

package com.datacrafts.digitalevers.simulationClick.common

case class sendData2Kafka(){}

当 actor 接受到自身的这个对象消息后，便会执行 case sendData2Kafka 内的代码，这里便可以实现向 Kafka 写数据的逻辑。

以上便是整个流程，如此简单。

Leave a comment