scala快速入门系列【Actor并发编程】

2021-01-26 16:02:05 浏览数 (1)

本篇作为scala快速入门系列的第三十八篇博客，为大家带来的是关于Actor并发编程的内容。

Actor并发编程

1.Actor介绍

scala的Actor并发编程模型可以用来开发比Java线程效率更高的并发程序。我们学习scala Actor的目的主要是为后续学习Akka做准备。

Java并发编程的问题

在Java并发编程中，每个对象都有一个逻辑监视器（monitor），可以用来控制对象的多线程访问。我们添加sychronized关键字来标记，需要进行同步加锁访问。这样，通过加锁的机制来确保同一时间只有一个线程访问共享数据。但这种方式存在资源争夺、以及死锁问题，程序越大问题越麻烦。

线程死锁

Actor并发编程模型

Actor并发编程模型，是scala提供给程序员的一种与Java并发编程完全不一样的并发编程模型，是一种基于事件模型的并发机制。Actor并发编程模型是一种不共享数据，依赖消息传递的一种并发编程模式，有效避免资源争夺、死锁等情况。

Java并发编程对比Actor并发编程

[NOTE]

  • scala在2.11.x版本中加入了Akka并发编程框架，老版本已经废弃。Actor的编程模型和Akka很像，我们这里学习Actor的目的是为学习Akka做准备。

2.创建Actor

创建Actor的方式和Java中创建线程很类似，也是通过继承来创建。

使用方式

  1. 定义class或object继承Actor特质
  2. 重写act方法
  3. 调用Actor的start方法执行Actor

[NOTE]

  • 类似于Java线程，这里的每个Actor是并行执行的

示例

创建两个Actor，一个Actor打印1-10，另一个Actor打印11-20

  • 使用class继承Actor创建（如果需要在程序中创建多个相同的Actor）
  • 使用object继承Actor创建（如果在程序中只创建一个Actor）

参考代码

使用class继承Actor创建

代码语言：javascript复制
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创建

代码语言：javascript复制
  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程序运行流程

  1. 调用start()方法启动Actor
  2. 自动执行act()方法
  3. 向Actor发送消息
  4. act方法执行完成后，程序会调用exit()方法

3.发送消息/接收消息

我们之前介绍Actor的时候，说过Actor是基于事件（消息）的并发编程模型，那么Actor是如何发送消息和接收消息的呢？

使用方式

发送消息

我们可以使用三种方法来发送消息：

例如:

要给actor1发送一个异步字符串消息，使用以下代码：

接收消息

Actor中使用receive方法来接收消息，需要给receive方法传入一个偏函数

[NOTE]

  • receive方法只接收一次消息，接收完后继续执行act方法

示例

  • 创建两个Actor（ActorSender、ActorReceiver）
  • ActorSender发送一个异步字符串消息给ActorReceiver
  • ActorReceive接收到该消息后，打印出来

参考代码

代码语言：javascript复制
  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()
  }

4.持续接收消息

通过上一个案例，ActorReceiver调用receive来接收消息，但接收一次后，Actor就退出了。

代码语言：javascript复制
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来接收消息就可以啦。

示例

  • 在上一个案例的基础上，让ActorReceiver能够一直接收消息
代码语言：javascript复制
  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()
  }

使用loop和react优化接收消息

上述代码，使用while循环来不断接收消息。

  • 如果当前Actor没有接收到消息，线程就会处于阻塞状态
  • 如果有很多的Actor，就有可能会导致很多线程都是处于阻塞状态
  • 每次有新的消息来时，重新创建线程来处理
  • 频繁的线程创建、销毁和切换，会影响运行效率

在scala中，可以使用loop react来复用线程。比while receive更高效

示例

  • 使用loop react重写上述案例

参考代码

代码语言：javascript复制
// 持续接收消息
loop {
    react {
        case msg:String => println("接收到消息："   msg)
    }
}

5.发送和接收自定义消息

我们前面发送的消息是字符串类型，Actor中也支持发送自定义消息，常见的如：使用样例类封装消息，然后进行发送处理。

示例一

  • 创建一个MsgActor，并向它发送一个同步消息，该消息包含两个字段（id、message）
  • MsgActor回复一个消息，该消息包含两个字段（message、name）
  • 打印回复消息

[!TIP]

  • 使用!?来发送同步消息
  • 在Actor的act方法中，可以使用sender获取发送者的Actor引用
代码语言：javascript复制
  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])
  }

示例二

  • 创建一个MsgActor，并向它发送一个异步无返回消息，该消息包含两个字段（message, company）

[!TIP]

  • 使用!发送异步无返回消息
代码语言：javascript复制
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("中国联通", "大爷，快交话费！")
}

示例三

  • 创建一个MsgActor，并向它发送一个异步有返回消息，该消息包含两个字段（id、message）
  • MsgActor回复一个消息，该消息包含两个字段（message、name）
  • 打印回复消息

[!TIP]

  • 使用!!发送异步有返回消息
  • 发送后，返回类型为Future[Any]的对象
  • Future表示异步返回数据的封装，虽获取到Future的返回值，但不一定有值，可能在将来某一时刻才会返回消息
  • Future的isSet()可检查是否已经收到返回消息，apply()方法可获取返回数据

参考代码

代码语言：javascript复制
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的经典案例，敬请期待٩(๑>◡<๑)۶

编程算法 java react scala

0 人点赞