本篇作为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创建

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()
  }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
 

使用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()
  }


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
 

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接收到该消息后，打印出来

参考代码

  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()
  }


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
 

4.持续接收消息

        通过上一个案例，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()
  }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
 

        上述代码，ActorReceiver无法接收到ActorSender发送的第二条消息。

        我们希望ActorReceiver能够一直接收消息，怎么实现呢？

        我们只需要使用一个while(true)循环，不停地调用receive来接收消息就可以啦。

示例

• 在上一个案例的基础上，让ActorReceiver能够一直接收消息

  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()
  }


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
 

使用loop和react优化接收消息

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

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

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

示例

• 使用loop + react重写上述案例

参考代码

// 持续接收消息
loop {
    react {
        case msg:String => println("接收到消息：" + msg)
    }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
 

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

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

示例一

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

[!TIP]

• 使用!?来发送同步消息
• 在Actor的act方法中，可以使用sender获取发送者的Actor引用

  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])
  }


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
 

示例二

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

[!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("中国联通", "大爷，快交话费！")
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
 

示例三

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

[!TIP]

• 使用!!发送异步有返回消息
• 发送后，返回类型为Future[Any]的对象
• Future表示异步返回数据的封装，虽获取到Future的返回值，但不一定有值，可能在将来某一时刻才会返回消息
• Future的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)
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
 

         本期的内容分享就到这里了,喜欢的小伙伴们记得点个赞，持续关注哟~下期为大家介绍如何使用Actor来完成WordCount的经典案例，敬请期待٩(๑>◡<๑)۶

文章来源: alice.blog.csdn.net，作者：大数据梦想家，版权归原作者所有，如需转载，请联系作者。

原文链接：alice.blog.csdn.net/article/details/104127034