Akka入门编程实践
Akka是使用Scala语言开发一个编程库，基于事件驱动的架构实现异步处理，它能够简化编写分布式应用程序。Akka中最核心的概念是Actor模型，它为编写分布式/并行计算应用程序提供了高层次抽象，在实际编程实践中，开发人员可以从对复杂网络通信细节的处理、多线程应用场景下对锁的管理中解脱出来。
Akka能够给应用程序带来的几个重要的特性是：
- 容错性
- 可伸缩性
- 异步性
- 事件驱动架构（EDA）
- 远程透明性
Actor是Akka中最核心的组件，以至于我们在编写基于Akka的应用程序时，大部分时间都会和Actor打交道，那么Actor到底是怎样的一种抽象呢？一个Actor对象封装了状态和行为，但是它不和外界其它的Actor共享状态，如果一个Actor想要和另一个Actor交互，能且只能通过发送消息来达到信息交换的目的。可见，一个Actor能够很好地保护其内部状态的安全。
与本地Actor通信
下面，我们从最简单的Actor编程来体验Akka的功能。首先，先定义几种类型的消息，后面会基于这些消息来进行通信，代码如下所示：
package org.shirdrn.scala.akka
object Start extends Serializable
object Stop extends Serializable
trait Message {
val id: String
}
case class Shutdown(waitSecs: Int) extends Serializable
case class Heartbeat(id: String, magic:Int) extends Message with Serializable
case class Header(id: String, len: Int, encrypted: Boolean) extends Message with Serializable
case class Packet(id: String, seq: Long, content: String) extends Message with Serializable
要实现一个Actor，需要继承自特质akka.actor.Actor，然后需要实现Actor特质声明的receive方法即可。另外，可选地可以混入另一个特质akka.actor.ActorLogging，提供记录日志的功能。我们首先实现的是一个Actor对象，然后拿到该Actor的一个引用（ActorRef），通过发送消息来与其进行交互，实现的Actor类为LocalActor ，代码如下所示：
class LocalActor extends Actor with ActorLogging {
def receive = {
case Start => log.info("start")
case Stop => log.info("stop")
case Heartbeat(id, magic) => log.info("Heartbeat" + (id, magic))
case Header(id, len, encrypted) => log.info("Header" + (id, len, encrypted))
case Packet(id, seq, content) => log.info("Packet" + (id, seq, content))
case _ =>
}
}
然后，实现一个带有main方法的类来与上面的LocalActor对象：
object LocalClient extends App {
// Local actor
val system= ActorSystem("local-system") // 创建一个ActorSystem对象，用来管理Actor实例
println(system )
val localActorRef =system.actorOf(Props(new LocalServer()), name="local-actor") // 通过ActorSystem对象，获取到一个Actor的引用
println(localActorRef)
localActorRef ! Start // 向LocalActor发送Start消息
localActorRef ! Heartbeat("3099100", 0xabcd) // 向LocalActor发送Heartbeat消息
// 创建一个JSON类型的消息Packet
val content = new JSONObject()
content.put("name", "Stone")
content.put("empid", 51082001)
content.put("score", 89.36581)
localActorRef ! Packet("3000001", System.currentTimeMillis(), content.toString) // 向LocalActor发送Packet消息
localActorRef ! Stop // 停止LocalActor实例
system shutdown // 终止ActorSystem对象，释放资源
}
虽然，我们只实现了一个本地Actor，但是这也非常有用，比如，我们在同一个JVM中有多个模块之间需要通过消息通信，完全可以实现多个本地Actor，他们之间进行通信，完成复杂的处理逻辑。
与远程Actor通信
在分布式应用场景中，通常需要跨节点进行通信，或者说交换消息，那么在使用Akka实现的时候就被抽象为在不同节点之上的多个Actor之间的交互。因为Akka提供的高层次抽象，所以在使用Akka编写分布式应用程序的时候，和编写本地应用程序一样简单。下面，我们实现一个伪分布式应用程序，使Actor在不同的JVM之内进行通信，实现上和在不同的节点上是一样的。
我们使用配置文件application.conf来指定通信处理过程中相关Actor的配置，包括远程Actor的主机名（或IP地址）和端口，包括本地Actor的基本配置。然后，只需要将该文件放在CLASSPATH之下即可，Akka会使用typesafe提供的配置解析工具ConfigFactory类来进行处理，配置文件application.conf中配置内容如下所示：
MyRemoteServerSideActor {
akka {
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
enabled-transports = ["akka.remote.netty.tcp"]
netty.tcp {
hostname = "127.0.0.1"
port = 2552
}
}
}
}
MyRemoteClientSideActor {
akka {
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
}
}
上面，MyRemoteServerSideActor指定了远程Actor的配置内容，Actor的provider配置为akka.remote.RemoteActorRefProvider，TPC通信配置的主机名为127.0.0.1，端口为2552；MyRemoteClientSideActor指定了本地Actor的配置，Actor的provider配置为akka.remote.RemoteActorRefProvider，下面看看代码实现。
- 远程Actor实现
实现远程Actor和实现一个本地Actor的方式是一样的，继承自特质Actor，并实现receive方法。我们实现的RemoteActor的代码如下所示：
class RemoteActor extends Actor with ActorLogging {
// 模拟处理结果状态，发送给消息的发送方
val SUCCESS = "SUCCESS"
val FAILURE = "FAILURE"
def receive = {
case Start => { // 处理Start消息
log.info("RECV event: " + Start)
}
case Stop => { // 处理Stop消息
log.info("RECV event: " + Stop)
}
case Shutdown(waitSecs) => { // 处理Shutdown消息
log.info("Wait to shutdown: waitSecs=" + waitSecs)
Thread.sleep(waitSecs)
log.info("Shutdown this system.")
context.system.shutdown // 停止当前ActorSystem系统
}
case Heartbeat(id, magic) => log.info("RECV heartbeat: " + (id, magic)) // 处理Heartbeat消息
case Header(id, len, encrypted) => log.info("RECV header: " + (id, len, encrypted)) // 处理Header消息
case Packet(id, seq, content) => { // 处理Packet消息
val originalSender = sender // 获取到当前发送方的Actor引用
log.info("RECV packet: " + (id, seq, content))
originalSender ! (seq, SUCCESS) // 响应给发送方消息处理结果，类似发送一个ACK
}
case _ =>
}
}
上面的Actor实现了接收多种类型的消息：Start、Stop、Shutdown、Heartbeat、Header、Packet，其中一个Shutdown消息是可以将当前远程ActorSystem系统终止的，终止后就无法再处理任何请求，而Packet消息则会给发送方一个返回，告知处理结果。
一个Actor可以在自己内部终止自己，需要通过执行context.system.shutdown就可以实现。
启动我们实现的远程Actor系统，等待接收并处理消息，如下所示：
object AkkaServerApplication extends App {
val system = ActorSystem("remote-system", ConfigFactory.load().getConfig("MyRemoteServerSideActor")) // 创建名称为remote-system的ActorSystem：从配置文件application.conf中获取该Actor的配置内容
val log = system.log
log.info("Remote server actor started: " + system)
system.actorOf(Props[RemoteActor], "remoteActor") // 创建一个名称为remoteActor的Actor，返回一个ActorRef，这里我们不需要使用这个返回值
}
这里是程序的主入口，启动改程序可以看到控制台输出如下内容：
[INFO] [08/14/2015 11:52:45.747] [main] [Remoting] Starting remoting [INFO] [08/14/2015 11:52:46.230] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://remote-system@127.0.0.1:2552] [INFO] [08/14/2015 11:52:46.232] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://remote-system@127.0.0.1:2552] [INFO] [08/14/2015 11:52:46.239] [main] [ActorSystem(remote-system)] Remote server actor started: akka://remote-system
可以看出，这与我们在配置文件，以及在代码中配置的内容相一致：ActorSystem系统名称为remote-system，通信端口为127.0.0.1:2552。
- 客户端Actor实现
我们再看本地将要与远程Actor通信的客户端Actor的实现，如下所示：
class ClientActor extends Actor with ActorLogging {
// akka.<protocol>://<actor system>@<hostname>:<port>/<actor path>
val path = "akka.tcp://remote-system@127.0.0.1:2552/user/remoteActor" // 远程Actor的路径，通过该路径能够获取到远程Actor的一个引用
val remoteServerRef = context.actorSelection(path) // 获取到远程Actor的一个引用，通过该引用可以向远程Actor发送消息
@volatile var connected = false
@volatile var stop = false
def receive = {
case Start => { // 发送Start消息表示要与远程Actor进行后续业务逻辑处理的通信，可以指示远程Actor初始化一些满足业务处理的操作或数据
send(Start)
if(!connected) {
connected = true
log.info("Actor connected: " + this)
}
}
case Stop => {
send(Stop)
stop = true
connected = false
}
case header: Header => send(header)
case hb: Heartbeat => sendWithCheck(hb)
case pkt: Packet => sendWithCheck(pkt)
case cmd: Shutdown => send(cmd)
case (seq, result) => log.info("RESULT: seq=" + seq + ", result=" + result) // 用于接收远程Actor处理一个Packet消息的结果
case m => log.info("Unknown message: " + m)
}
private def sendWithCheck(cmd: Serializable): Unit = {
while(!connected) {
Thread.sleep(100)
log.info("Wait to be connected...")
}
if(!stop) {
send(cmd)
} else {
log.warning("Actor has stopped!")
}
}
private def send(cmd: Serializable): Unit = {
log.info("Send command to server: " + cmd)
try {
remoteServerRef ! cmd // 发送一个消息到远程Actor，消息必须是可序列化的，因为消息对象要经过网络传输
} catch {
case e: Exception => {
connected = false
log.info("Try to connect by sending Start command...")
send(Start)
}
}
}
}
本地Actor会接收处理本地（当前JVM中）发送过来的消息，一个简单的check，然后进行转发，发送到远程Actor；也用来接收来自远程Actor响应的处理结果。接收并转发本地消息，包括如下类型消息：Start、Stop、Shutdown、Header、Heartbeat、Packet。其中，我们会在本地客户端创建一个单独的线程去周期性地发送心跳消息Heartbeat到远程Actor，同时将大量的Packet消息发送到远程Actor去处理。接收到的远程Actor响应的消息是一个Tuple类型，可以提取出seq和result数据，查看某个消息处理结果。下面是本地客户端的实现逻辑，如下所示：
object AkkaClientApplication extends App {
val system = ActorSystem("client-system", ConfigFactory.load().getConfig("MyRemoteClientSideActor")) // 通过配置文件application.conf配置创建ActorSystem系统
val log = system.log
val clientActor = system.actorOf(Props[ClientActor], "clientActor") // 获取到ClientActor的一个引用
@volatile var running = true
val hbInterval = 1000
lazy val hbWorker = createHBWorker
/**
* create heartbeat worker thread
*/
def createHBWorker: Thread = { // 心跳发送线程
new Thread("HB-WORKER") {
override def run(): Unit = {
while(running) {
clientActor ! Heartbeat("HB", 39264)
Thread.sleep(hbInterval)
}
}
}
}
def format(timestamp: Long, format: String): String = {
val df = new SimpleDateFormat(format)
df.format(new Date(timestamp))
}
def createPacket(packet: Map[String, _]): JSONObject = {
val pkt = new JSONObject()
packet.foreach(p => pkt.put(p._1, p._2))
pkt
}
val ID = new AtomicLong(90760000)
def nextTxID: Long = {
ID.incrementAndGet()
}
clientActor ! Start // 发送一个Start消息，第一次与远程Actor握手（通过本地ClientActor进行转发）
Thread.sleep(2000)
clientActor ! Header("HEADER", 20, encrypted=false) // 发送一个Header消息到远程Actor（通过本地ClientActor进行转发）
Thread.sleep(2000)
hbWorker.start // 启动心跳线程
// send some packets
val DT_FORMAT = "yyyy-MM-dd HH:mm:ss.SSS"
val r = Random
val packetCount = 100
val serviceProviders = Seq("CMCC", "AKBBC", "OLE")
val payServiceProvicers = Seq("PayPal", "CMB", "ICBC", "ZMB", "XXB")
def nextProvider(seq: Seq[String]): String = {
seq(r.nextInt(seq.size))
}
val startWhen = System.currentTimeMillis()
for(i <- 0 until packetCount) { // 持续发送packetCount个Packet消息
val pkt = createPacket(Map[String, Any](
"txid" -> nextTxID,
"pvid" -> nextProvider(serviceProviders),
"txtm" -> format(System.currentTimeMillis(), DT_FORMAT),
"payp" -> nextProvider(payServiceProvicers),
"amount" -> 1000 * r.nextFloat()))
clientActor ! Packet("PKT", System.currentTimeMillis, pkt.toString)
}
val finishWhen = System.currentTimeMillis()
log.info("FINISH: timeTaken=" + (finishWhen - startWhen) + ", avg=" + packetCount/(finishWhen - startWhen))
Thread.sleep(2000)
// ask remote actor to shutdown
val waitSecs = hbInterval
clientActor ! Shutdown(waitSecs) // 发送Packet消息完成，通知远程Actor终止服务
running = false
while(hbWorker.isAlive) { // 终止心跳线程
log.info("Wait heartbeat worker to exit...")
Thread.sleep(300)
}
system.shutdown // 终止本地ActorSystem系统
}
上面代码中有详细注释，可以了解具体实现。
使用Akka Future实例
前面的两种情况，我们模拟了Actor如果在本地/远程的上下文中进行通信处理，Akka很好地屏蔽了底层网络通信细节。接下来我们看看看Akka的Future功能，尤其是Future所支持异步Callback特性。
我们基于Akka实现的例子，如下图所示：
上图模拟了一个简易的有趣的爬虫系统，而且在这上面为了演示Akka的使用，我们在各个Actor之间增加了好多消息通信，可以根据上图中箭线上的编号来理解整个实例系统的执行流程。
存储网页链接，以及一个指定网页的出链接（Outlink）信息，我们使用MySQL数据库，创建了2个数据表，数据库及其表定义如下所示：
GRANT ALL ON *.* TO 'web'@'%' IDENTIFIED BY 'web'; CREATE DATABASE `page_db` DEFAULT CHARACTER SET utf8; -- 用来存储一个链接，以及该链接对应的页面的相关信息 CREATE TABLE `web_link` ( `link` CHAR(128) NOT NULL UNIQUE, `domain` VARCHAR(64) NOT NULL, `encoding` VARCHAR(11) NOT NULL DEFAULT 'utf-8', `content_length` INT NOT NULL DEFAULT 0, `create_time` VARCHAR(20) NOT NULL, PRIMARY KEY (`link`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; -- 用来存储一个链接的outlink信息 CREATE TABLE `web_outlink` ( `link` CHAR(128) NOT NULL, `outlink` VARCHAR(128) NOT NULL, `create_time` VARCHAR(20) NOT NULL, PRIMARY KEY (`link`, `outlink`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
另外，Akka中Actor之间通过发送消息进行通信，所以我们首先定义几个case class，如下所示：
case class WebUrl(link: String) case class ScheduledWebUrl(link: String, config: Map[String, Any]) case class CrawledWeb(link: String, domain: String, encoding: String, contentLength: Int, outlinks: Set[String]) case class Stored(link: String, outlinkCount: Int)
还有，操作MySQL以及日期时间转换操作，我们实现了两个工具类，如下所示：
object MySQLUtils {
val driverClass = "com.mysql.jdbc.Driver"
val jdbcUrl = "jdbc:mysql://10.10.4.130:3306/page_db"
val user = "web"
val password = "web"
try {
Class.forName(driverClass)
} catch {
case e: ClassNotFoundException => throw e
case e: Exception => throw e
}
@throws(classOf[SQLException])
def getConnection: Connection = {
DriverManager.getConnection(jdbcUrl, user, password)
}
@throws(classOf[SQLException])
def doTrancation(transactions: Set[String]) : Unit = {
val connection = getConnection
connection.setAutoCommit(false)
transactions.foreach {
connection.createStatement.execute(_)
}
connection.commit
connection.close
}
}
object DatetimeUtils {
val DEFAULT_DT_FORMAT = "yyyy-MM-dd HH:mm:ss"
def format(timestamp: Long, format: String): String = {
val df = new SimpleDateFormat(format)
df.format(new Date(timestamp))
}
def format(timestamp: Long): String = {
val df = new SimpleDateFormat(DEFAULT_DT_FORMAT)
df.format(new Date(timestamp))
}
}
下面详细上图中各个组件的实现：
- 程序入口AkkaCrawlApp
AkkaCrawlApp是程序入口，读取Seed链接信息，然后发送给CrawlActor，使整个系统开始运行起来。AkkaCrawlApp代码实现如下所示：
object AkkaCrawlApp {
def main(args: Array[String]) {
val system = ActorSystem("crawler-system") // 创建一个ActorSystem
system.log.info(system.toString)
val scheduleActorRef = system.actorOf(Props[ScheduleActor], name="schedule-actor") // 创建ScheduleActor
val storeActorRef = system.actorOf(Props[PageStoreActor], name="store-actor") // 创建PageStoreActor
val crawlActorRef = system.actorOf(Props[CrawlActor], name="crawl-actor") // 创建CrawlActor
val links =
"""
|http://apache.org
|http://csdn.net
|http://hadoop.apache.org
|http://spark.apache.org
|http://nutch.apache.org
|http://storm.apache.org
|http://mahout.apache.org
|http://flink.apache.org
|http://fdajlkdjfakfjlkadjflkajflakjf.com
""".stripMargin
val seeds: Seq[String] = links.split("\\s+").toSeq
ScheduleActor.sendFeeds(crawlActorRef, seeds) // 调用ScheduleActor的伴生对象的sendFeeds，将爬虫入口seed链接发送给CrawlActor
}
}
我们在这里创建了一个ActorSystem实例，然后该系统有3个Actor实例：ScheduleActor 、PageStoreActor、CrawlActor。如果是在该ActorSystem上下文中，可以通过Actor的名称检索到该Actor的引用ActorRef实例，然后通过向该引用发送消息来进行通信。
- ScheduleActor伴生对象
ScheduleActor伴生对象类中，通过传入一个CrawlActor的引用，将传入的Seed链接发送给该系统中的CrawlActor。ScheduleActor伴生对象类代码如下所示：
object ScheduleActor {
def sendFeeds(crawlerActorRef: ActorRef, seeds: Seq[String]): Unit = {
seeds.foreach(crawlerActorRef ! _)
}
}
上面代码比较简单，遍历seed链接集合，向CrawlActor发送传入的每一个Seed链接。
- CrawlActor实现
CrawlActor是一个Actor，它用来处理链接信息，同时也负责下载网页数据并抽取出链接（Outlink），代码实现如下所示：
class CrawlActor extends Actor with ActorLogging {
// 获取到ScheduleActor和PageStoreActor的引用
private val scheduleActor = context.actorOf(Props[ScheduleActor], "schedule_actor")
private val storeActor = context.actorOf(Props[PageStoreActor], "store_actor")
private val q = new LinkedBlockingQueue[String]()
implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(new ForkJoinPool())
def receive = {
case link: String => { // 系统启动时，ScheduleActor伴生对象将入口seed链接字符串发过来
if(link != null && link.startsWith("http://")) { // 简单验证链接合法性
log.info("Checked: " + link)
scheduleActor ! WebUrl(link) // 将字符串链接link包装成为WebUrl发送回ScheduleActor，等待进一步调度
}
}
case ScheduledWebUrl(link, _) => { // ScheduleActor将包含调度信息和链接信息的链接任务发过来，指派CrawlActor下载网页内容
var crawledWeb: CrawledWeb = null
val crawlFuture = Future { // 创建一个Future，里面的逻辑包含了下载网页的代码
try {
var encoding = "utf-8"
var outlinks: Set[String] = Set[String]()
val u = new URL(link)
val domain = u.getHost
val uc = u.openConnection().asInstanceOf[HttpURLConnection]
uc.setConnectTimeout(5000)
uc.connect()
if (uc.getResponseCode == 200) {
// page encoding
if (uc.getContentEncoding != null) {
encoding = uc.getContentEncoding
}
// page content
if (uc.getContentLength > 0) {
val in = uc.getInputStream
val buffer = Array.fill[Byte](512)(0)
val baos = new ByteArrayOutputStream
var bytesRead = in.read(buffer)
while (bytesRead > -1) {
baos.write(buffer, 0, bytesRead)
bytesRead = in.read(buffer)
}
outlinks = extractOutlinks(link, baos.toString(encoding)) // 抽取网页中的出链接
baos.close
}
log.info("Page: link=" + link + ", encoding=" + encoding + ", outlinks=" + outlinks)
CrawledWeb(link, domain, encoding, uc.getContentLength, outlinks)
}
} catch {
case e: Throwable => {
log.error("Crawl error: " + e.toString)
e
}
}
}
// 这里设置了一个回调，当下载网页的Future完成后，会调用这里的回调方法
crawlFuture.onSuccess { // 下载成功，则Future返回CrawledWeb对象
case crawledWeb: CrawledWeb => {
log.info("Succeed to crawl: link=" + link + ", crawledWeb=" + crawledWeb)
if(crawledWeb != null) {
storeActor ! crawledWeb // 通知PageStoreActor保存网页相关信息
log.info("Sent crawled data to store actor.")
q add link // 将链接缓存到队列中
}
}
}
crawlFuture.onFailure { // 下载失败，则Future返回Throwable异常对象，这里简单打印出异常内容
case exception: Throwable => log.error("Fail to crawl: " + exception.toString)
}
}
case Stored(link, count) => { // 当PageStoreActor保存网页信息成功，会给一个Stored通知
q.remove(link)
scheduleActor ! (link, count) // 向ScheduleActor汇总统计结果
}
}
def extractOutlinks(parentUrl: String, content: String): Set[String] = { // 使用正则表达式抽取页面上的链接
val outlinks = "href\\s*=\\s*\"([^\"]+)\"".r.findAllMatchIn(content).map { m =>
var url = m.group(1)
if(!url.startsWith("http")) {
url = new URL(new URL(parentUrl), url).toExternalForm
}
url
}.toSet
// 只保留页面上以html和htm结尾的链接
outlinks.filter( url => !url.isEmpty && (url.endsWith("html") || url.endsWith("htm")))
}
}
这里需要说明的是，在使用Future的回调功能时，有两种方式：
一种是使用onSuccess和onFailure方法，这就要在Future执行代码段中设置好处理逻辑执行成功时，返回值是什么类型，可能有多重成功的情况，如下载页面成功、由于没使用代理无法下载也算成功结束，然后在onSuccess方法中通过case来分别处理；如果Future中代码处理失败，则在onFailure方法中处理异常，例如如果是因为没使用代理无法访问页面，可以在这里选择一个代理地址，然后重新发回给ScheduleActor进行调度下载。
另一种是使用onComplete方法来处理回调，这里面能够同时处理Future代码执行成功或者失败的情况，根据自己的习惯选择使用。
上面代码中，我们使用了下面一个implicit定义：
implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(new ForkJoinPool())
这里指定了一个隐式ExecutionContext变量ec，它提供了一个线程池来执行Future中定义的任务，也可以看看Future的onSuccess方法，它是一个柯里化（Currying）函数，需要传入一个ExecutionContext，如下所示：
def onSuccess[U](pf: PartialFunction[T, U])(implicit executor: ExecutionContext): Unit = onComplete {
case Success(v) =>
pf.applyOrElse[T, Any](v, Predef.conforms[T]) // Exploiting the cached function to avoid MatchError
case _ =>
}
如果每次调用都显式传入一个ExecutionContext实例，代码看起来会非常丑陋，所以通过定义隐式ExecutionContext，使代码看起来更直观清晰。通过上面的onSuccess方法的定义，我们也能看到，它内部实际上调用了onComplete方法。
- PageStoreActor实现
PageStoreActor主要用来将数据进行持久化，它会直接操作MySQL数据库，代码如下所示：
class PageStoreActor extends Actor with ActorLogging {
implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(new ForkJoinPool())
var crawlerRef = context.actorOf(Props[CrawlActor], name="crawl-actor")
def receive = {
case CrawledWeb(link, domain, encoding, contentLength, outlinks) => {
val future = Future {
var sqls = Set[String]()
try {
val createTime = DatetimeUtils.format(System.currentTimeMillis)
val sql = "INSERT INTO web_link VALUES ('" + link + "','" + domain + "','" + encoding + "'," + contentLength + ",'" + createTime + "')"
log.info("Link SQL: " + sql)
sqls += sql
var outlinksSql = "INSERT INTO web_outlink VALUES "
outlinksSql += outlinks.map("('" + link + "','" + _ + "','" + createTime + "')").mkString(",")
log.info("Outlinks SQL: " + outlinksSql)
sqls += outlinksSql
// 使用了事务操作
MySQLUtils.doTrancation(sqls)
(link, outlinks.size)
} catch {
case e: Throwable => throw e
}
}
// 这里也使用了Future的回调功能
future.onSuccess {
case (link: String, outlinkCount: Int) => {
log.info("SUCCESS: link=" + link + ", outlinkCount=" + outlinkCount)
crawlerRef ! Stored(link, outlinkCount) // 将持久化后的链接及其outlink数量回馈给CrawlActor
}
}
future.onFailure {
case e: Throwable => throw e
}
}
}
}
上面代码比较容易理解。
- ScheduleActor实现
ScheduleActor负责调度任务，将带有调度信息的链接发送给CrawlActor去下载链接对应的页面。同时，页面下载、保存完成以后，ScheduleActor会收集结果信息，保存在内部的一个ConcurrentHashMap中，实现代码如下所示：
class ScheduleActor extends Actor with ActorLogging {
val config = Map(
"domain.black.list" -> Seq("google.com", "facebook.com", "twitter.com"),
"crawl.retry.times" -> 3,
"filter.page.url.suffixes" -> Seq(".zip", ".avi", ".mkv")
)
val counter = new ConcurrentHashMap[String, Int]()
def receive = {
case WebUrl(url) => {
sender ! ScheduledWebUrl(url, config)
}
case (link: String, count: Int) => {
counter.put(link, count)
log.info("Counter: " + counter.toString)
}
}
}
- Akka应用扩展
上面的例子，我们是在单机上运行的，起始可以很容易将其扩展到多机环境。结合上面我们的Akka Remoting实践，可以将各个Actor分别运行在不同的进程之内（单机多进程，或跨机器多进程），通过配置文件的方式，单独搞一个object工具类用来解析配置文件，根据实际需要创建所需要通信的Actor（ActorRef或ActorSelection），其他其他代码的处理几乎不需要做改动，就能运行。
另外，也可以使用Akka提供的内置Routing策略，来实现消息的路由。