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

网友投稿 798 2022-05-30

本篇作为Scala快速入门系列的第三十八篇博客,为大家带来的是关于

Actor并发编程

的内容。

文章目录

Actor并发编程

1.Actor介绍

Java并发编程的问题

Actor并发编程模型

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

2.创建Actor

使用方式

示例

Actor程序运行流程

3.发送消息/接收消息

使用方式

示例

4.持续接收消息

示例

使用loop和react优化接收消息

示例

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

示例一

示例二

示例三

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中创建线程很类似,也是通过继承来创建。

使用方式

定义class或object继承Actor特质

重写act方法

调用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

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

9

10

11

12

13

14

15

16

17

18

19

Actor程序运行流程

调用start()方法启动Actor

自动执行act()方法

向Actor发送消息

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

的经典案例,敬请期待٩(๑>◡<๑)۶

Java Scala

版权声明:本文内容由网络用户投稿,版权归原作者所有,本站不拥有其著作权,亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容,请联系我们jiasou666@gmail.com 处理,核实后本网站将在24小时内删除侵权内容。

上一篇:ADG单实例搭建系列之(Active Database Duplicate Using Image Copies)
下一篇:性能工具之 Locust 工具关联与参数化
相关文章