本篇作为scala快速入门系列的第三十八篇博客,为大家带来的是关于Actor并发编程的内容。
scala的Actor并发编程模型可以用来开发比Java线程效率更高的并发程序。我们学习scala Actor的目的主要是为后续学习Akka做准备。
在Java并发编程中,每个对象都有一个逻辑监视器(monitor),可以用来控制对象的多线程访问。我们添加sychronized关键字来标记,需要进行同步加锁访问。这样,通过加锁的机制来确保同一时间只有一个线程访问共享数据。但这种方式存在资源争夺、以及死锁问题,程序越大问题越麻烦。
线程死锁
Actor并发编程模型,是scala提供给程序员的一种与Java并发编程完全不一样的并发编程模型,是一种基于事件模型的并发机制。Actor并发编程模型是一种不共享数据,依赖消息传递的一种并发编程模式,有效避免资源争夺、死锁等情况。
[NOTE]
创建Actor的方式和Java中创建线程很类似,也是通过继承来创建。
[NOTE]
创建两个Actor,一个Actor打印1-10,另一个Actor打印11-20
参考代码
使用class继承Actor创建
object _05ActorDemo {
class Actor1 extends Actor {
override def act(): Unit = (1 to 10).foreach(println(_))
}
class Actor2 extends Actor {
override def act(): Unit = (11 to 20).foreach(println(_))
}
def main(args: Array[String]): Unit = {
new Actor1().start()
new Actor2().start()
}
}
使用object继承Actor创建
object Actor1 extends Actor {
override def act(): Unit =
for(i <- 1 to 10) {
println(i)
}
}
object Actor2 extends Actor {
override def act(): Unit =
for(i <- 11 to 20) {
println(i)
}
}
def main(args: Array[String]): Unit = {
Actor1.start()
Actor2.start()
}
我们之前介绍Actor的时候,说过Actor是基于事件(消息)的并发编程模型,那么Actor是如何发送消息和接收消息的呢?
发送消息
我们可以使用三种方法来发送消息:
例如:
要给actor1发送一个异步字符串消息,使用以下代码:
接收消息
Actor中使用receive方法来接收消息,需要给receive方法传入一个偏函数
[NOTE]
参考代码
object ActorSender extends Actor {
override def act(): Unit = {
// 发送消息
while(true) {
ActorReceiver ! "hello!"
TimeUnit.SECONDS.sleep(3)
}
}
}
object ActorReceiver extends Actor {
override def act(): Unit = {
// 持续接收消息
while(true) {
receive {
case msg:String => println("接收到消息:" + msg)
}
}
}
}
def main(args: Array[String]): Unit = {
ActorReceiver.start()
ActorSender.start()
}
通过上一个案例,ActorReceiver调用receive来接收消息,但接收一次后,Actor就退出了。
object ActorSender extends Actor {
override def act(): Unit = {
// 发送字符串消息给Actor2
val msg = "你好,ActorSender"
println(s"ActorSender: 发送消息$msg")
ActorReceiver ! msg
// 再次发送一条消息,ActorReceiver无法接收到
ActorReceiver ! "你叫什么名字?"
}
}
object ActorReceiver extends Actor {
override def act(): Unit =
receive {
case msg: String => println(s"接收Actor: 接收到$msg")
}
}
object ActorMsgDemo {
def main(args: Array[String]): Unit = {
ActorSender.start()
ActorReceiver.start()
}
}
上述代码,ActorReceiver无法接收到ActorSender发送的第二条消息。
我们希望ActorReceiver能够一直接收消息,怎么实现呢?
我们只需要使用一个while(true)循环,不停地调用receive来接收消息就可以啦。
object ActorSender extends Actor {
override def act(): Unit = {
// 发送消息
while(true) {
ActorReceiver ! "hello!"
TimeUnit.SECONDS.sleep(3)
}
}
}
object ActorReceiver extends Actor {
override def act(): Unit = {
// 持续接收消息
while(true) {
receive {
case msg:String => println("接收到消息:" + msg)
}
}
}
}
def main(args: Array[String]): Unit = {
ActorReceiver.start()
ActorSender.start()
}
上述代码,使用while循环来不断接收消息。
在scala中,可以使用loop + react来复用线程。比while + receive更高效
参考代码
// 持续接收消息
loop {
react {
case msg:String => println("接收到消息:" + msg)
}
}
我们前面发送的消息是字符串类型,Actor中也支持发送自定义消息,常见的如:使用样例类封装消息,然后进行发送处理。
[!TIP]
!?
来发送同步消息 case class Message(id:Int, msg:String)
case class ReplyMessage(msg:String, name:String)
object MsgActor extends Actor {
override def act(): Unit = {
loop {
react {
case Message(id, msg) => {
println(s"接收到消息:${id}/${msg}")
sender ! ReplyMessage("不太好", "Tom")
}
}
}
}
}
def main(args: Array[String]): Unit = {
MsgActor.start()
val replyMessage: Any = MsgActor !? Message(1, "你好")
println("回复消息:" + replyMessage.asInstanceOf[ReplyMessage])
}
[!TIP]
!
发送异步无返回消息case class Mesasge(message:String, company:String)
object MsgActor extends Actor {
override def act(): Unit = {
loop {
react {
case Mesasge(message, company) =>
println(s"MsgActor接收到消息:${message}/${company}")
}
}
}
}
def main(args: Array[String]): Unit = {
MsgActor.start()
MsgActor ! Mesasge("中国联通", "大爷,快交话费!")
}
[!TIP]
!!
发送异步有返回消息Future[Any]
的对象isSet()可检查是否已经收到返回消息,apply()方法可获取返回数据
参考代码
case class Message(id:Int, message:String)
case class ReplyMessage(message:String, name:String)
object MsgActor extends Actor {
override def act(): Unit = {
loop {
react {
case Message(id, message) =>
println(s"MsgActor接收到消息:${id}/${message}")
sender ! ReplyMessage("收到消息!", "JIm")
}
}
}
}
def main(args: Array[String]): Unit = {
MsgActor.start()
val future: Future[Any] = MsgActor !! Message(1, "你好!")
while(!future.isSet) {}
val replyMessage = future.apply().asInstanceOf[ReplyMessage]
println(replyMessage)
}
本期的内容分享就到这里了,喜欢的小伙伴们记得点个赞,持续关注哟~下期为大家介绍如何使用Actor来完成WordCount的经典案例,敬请期待?(?>?<?)?