Scala Actor

二、什么是Scala Actor

1．概念
Scala中的Actor能够实现并行编程的强大功能，它是基于事件模型的并发机制，Scala是运用消息（message）的发送、接收来实现多线程的。使用Scala能够更容易地实现多线程应用的开发。
2．传统java并发编程与Scala Actor编程的区别
Scala Actor,receive不断接收消息，react复用线程，结合case class的actor,Future使用,使用Actor进行wordCount

对于Java，我们都知道它的多线程实现需要对共享资源（变量、对象等）使用synchronized 关键字进行代码块同步、对象锁互斥等等。而且，常常一大块的try…catch语句块中加上wait方法、notify方法、notifyAll方法是让人很头疼的。原因就在于Java中多数使用的是可变状态的对象资源，对这些资源进行共享来实现多线程编程的话，控制好资源竞争与防止对象状态被意外修改是非常重要的，而对象状态的不变性也是较难以保证的。而在Scala中，我们可以通过复制不可变状态的资源（即对象，Scala中一切都是对象，连函数、方法也是）的一个副本，再基于Actor的消息发送、接收机制进行并行编程
3． Actor方法执行顺序
1.首先调用start()方法启动Actor
2.调用start()方法后其act()方法会被执行
Scala Actor向Actor发送消息
4．发送消息的方式
! 发送异步消息，没有返回值。
!? 发送同步消息，等待返回值。
!! 发送异步消息，返回值是 Future[Any]。

三、 Actor实战

如果发现Actor的包引入不了，解决办法是：
选中项目右键—>Open Module Settings—>Library–>添加jar(scala-2.10.6.zip中的\lib*.jar)
上面的右侧就是我们引入的jar包
Scala Actor,receive不断接收消息，react复用线程，结合case class的actor,Future使用,使用Actor进行wordCount
1．第一个例子

package cn.toto.scala.day2
import scala.actors.Actor

//这种执行完成之后就停止了。
object MyActor1 extends Actor {
//重新act方法
def act(): Unit = {
for(i <- 1 to 10) {
      println("actor-1 " + i)
      Thread.sleep(3000)
    }
  }
}

object MyActor2 extends Actor {

def act(): Unit = {
for(i <- 1 to 10) {
      println("actor-2 " + i)
      Thread.sleep(2500)
    }
  }
}

object ActorTest extends App {
//因为是object类型的，不需要new,可以直接调用start方法
  MyActor1.start()
  MyActor2.start()
}

说明：上面分别调用了两个单例对象的start()方法，他们的act()方法会被执行，相同与在java中开启了两个线程，线程的run()方法会被执行
注意：这两个Actor是并行执行的，act()方法中的for循环执行完成后actor程序就退出了

2．第二个例子（可以不断地接收消息）
receive相当于是创建线程和销毁线程的过程。

package cn.toto.scala.day2

import scala.actors.Actor

/**
 * Created by toto on 2017/7/1.
 */
class MyActor extends Actor {
override def act(): Unit = {
while(true) {
      receive {
case "start" => {
          println("starting...")
          Thread.sleep(5000)
          println("stared")
        }
case "stop" => {
          println("stoping ...")
          Thread.sleep(5000)
          println("stopped ...")
        }
      }
    }
  }
}

//伴生对象，说明在一个Actor内部是串行的，如果想是并发执行的，可以创建多个actor
object MyActor {
def main(args: Array[String]): Unit = {
val actor = new MyActor
    actor.start()
//发送异步消息，感叹号就相当于是一个方法，可以在感叹号后面的字符串两边加载上括号
    actor ! "start"
    actor ! "stop"
    println("xiao ")
  }
}

说明：在act()方法中加入了while (true) 循环，就可以不停的接收消息
注意：发送start消息和stop的消息是异步的，但是Actor接收到消息执行的过程是同步的按顺序执行

3．第三个例子（react方式会复用线程，比receive更高效）
react类似线程池机制，可以复用线程。

package cn.toto.scala.day2

import scala.actors.Actor

/**
 * Created by toto on 2017/7/1.
 */
class YourActor extends Actor {
override def act(): Unit = {
    loop {
//这里是一个偏函数
      react {
case "start" => {
          println("starting...")
          Thread.sleep(5000)
          println("started")
        }
case "stop" => {
          println("stopping...")
          Thread.sleep(8000)
          println("stopping")
        }
      }
    }
  }
}

/**
 * 运行结果：
 * 消息发送完成!
 * starting...
 * started
 * stopping...
 * stopping
 */
object YourActor {
def main(args: Array[String]): Unit = {
val actor = new YourActor
    actor.start()
    actor ! "start"
    actor ! "stop"
    println("消息发送完成!")
  }
}

运行结果是：
消息发送完成!
starting...
started
stopping...
stopping...

说明： react 如果要反复执行消息处理，react外层要用loop，不能用while

4．第四个例子（结合case class发送消息）

package cn.toto.scala.day2

import scala.actors.Actor

case class SyncMsg(id:Int,msg:String)
case class AsyncMsg(id:Int,msg:String)
case class ReplyMsg(id:Int,msg:String)

/**
 * Created by toto on 2017/7/1.
 */
class AppleActor extends Actor {

def act(): Unit = {
while (true) {
      receive {
case "start" => println("starting ...")
case SyncMsg(id, msg) => {
          println(id + ",sync " + msg)
          Thread.sleep(5000)
//这里又开了一个子线程，这个在下面的方法中执行
//sender也是一个方法。看可以通过消息的发送者发送
          sender ! ReplyMsg(3,"finished")
        }
case AsyncMsg(id, msg) => {
          println(id + ",async " + msg)
          Thread.sleep(5000)
        }
      }
    }
  }
}

object AppleActor {

/**
 * 总体的运行结果是：
 * 异步消息发送完成
 * 1,async hello actor
 * None
 * false
 * 123
 * 2,sync hello actor
 * 2,sync hello actor
 * true
 * ReplyMsg(3,finished)
 * @param args
 */
def main(args: Array[String]): Unit = {
val a = new AppleActor
    a.start()
//异步消息
    a ! AsyncMsg(1,"hello actor")
    println("异步消息发送完成")

//同步消息(因为这里的最长时间是1秒，但是SyncMsg中休眠的时间更长，为5秒，所以这里是None)
//同步消息，表示需要等待
val content = a.!? (1000,SyncMsg(2,"hello actor"))
    println(content)

//异步但是返回结果，这个结果会被放到future中。然后进行返回。
val reply = a !! SyncMsg(2,"hello actor")
//replay是否已经完成通过isSet类判断
    println(reply.isSet);

    println("123")

//通过apply方法拿到里面的值，这个apply是feture中的方法。
val c = reply.apply()
//拿到之后，reply中的值相当于被设置了。所以c为true
    println(reply.isSet)
    println(c)
  }
}

Future的使用

package cn.toto.scala.day2;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import java.util.concurrent.*;

/**
 * Created by ZhaoXing on 2016/6/27.
 * Feture相当于给出了一种期望，就是说可能立即不会有结果，但是执行完毕或过一段时间后会有结果
 */
public class FutureDemo {

/**
 * 运行结果：
 * main false
 * 正在查找数据
 * pool-1-thread-1 数据查找完毕
 * true
 * 9527
 * @param args
 * @throws Exception
 */
public static void main(String[] args) throws Exception{
        ExecutorService threadPool = Executors.newCachedThreadPool();

        Future<Integer> future = threadPool.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
                System.out.println("正在查找数据");
                Thread.sleep(4000);
                System.out.println(Thread.currentThread().getName() + " 数据查找完毕");
return 9527;
            }
        });


        System.out.println(Thread.currentThread().getName() + " "  + future.isDone());
//System.out.println(future.get(1000, TimeUnit.MILLISECONDS));
        Thread.sleep(6000);
        System.out.println(future.isDone());
        System.out.println(future.get());
    }
}

使用Actor进行wordCount

package cn.toto.scala.day2

import scala.actors.{Actor, Future}
import scala.collection.mutable.ListBuffer
import scala.collection.parallel.Task
import scala.io.Source

/**
 * Created by toto on 2017/7/1.
 *
 * 使用线程的方式计算单词出现的频率
 */
case class SbTask(fn:String)
case object StopTask

class WordCountByActorTask extends Actor {
override def act(): Unit = {
while(true) {
      receive {
case SbTask(f) => {
//通过Source.fromFile(f)读取文件，获取每行并且把它转换成List
val lines : List[String] = Source.fromFile(f).getLines().toList;
//将lines list里面的内容合并，然后并且将它split.
val words : List[String] = lines.flatMap(_.split(" "))
//将word这些单词变成map的并且是元组类型的，每个都是1，----,记着对他们进行过分组，接着就是计算单词
val result: Map[String, Int]= words.map((_, 1)).groupBy(_._1).mapValues(_.size)
          sender ! result
        }
case StopTask => {
          exit()
        }
      }
    }
  }
}

object WordCountByActor {
def main(args: Array[String]): Unit = {
val replys = new ListBuffer[Future[Any]]
val results = new ListBuffer[Map[String,Int]]
val files = Array("E://wordcount//input//a.txt","E://wordcount//input//b.txt","E://wordcount//input//c.txt")
//每个文件就启动多少个actor,并将每个actor处理后得到的结果存储到List中
for(f <- files) {
val t = new WordCountByActorTask
      t.start()
//通过SbTask的方式发送文件名
val reply:Future[Any] = t !! SbTask(f)
//将处理到的结果放到ListBuffer中。
      replys += reply
    }

//对获得的replay的值进行计算。
while (replys.size > 0) {
val dones: ListBuffer[Future[Any]] = replys.filter(_.isSet)
for(f <- dones) {
//从relay中拿值，并将它转成map
        results += f.apply().asInstanceOf[Map[String, Int]]
//计算完成之后，将这个replay移除掉。
        replys -= f
      }
      Thread.sleep(500)
    }

    println(results.flatten.groupBy(_._1).mapValues(_.foldLeft(0)(_+_._2)))
  }
}

秒客网

Scala Actor,receive不断接收消息，react复用线程，结合case class的actor,Future使用,使用Actor进行wordCount

二、什么是Scala Actor

三、 Actor实战

相关文章

Scala Actor,receive不断接收消息，react复用线程，结合case class的actor,Future使用,使用Actor进行wordCount

二、 什么是Scala Actor

三、 Actor实战

相关文章

二、什么是Scala Actor