讨喜的隔离可变性(四)收发消息

简介:

声明:本文是《Java虚拟机并发编程》的第五章,感谢华章出版社授权并发编程网站发布此文,禁止以任何形式转载此文。

我们可以向角色发送任何类型的消息——String、Integer、Long、Double、List、Map、元组(tuples)、Scala的case类…但其中有一点需要注意的是,上述所有类型的消息都必须是不可变的。在上述这些类型中,我对于元组有着特殊的偏好,这并非因为我听到别人把元组误读成“two-ples”时感到很有趣,而是由于元组是轻量的、不可变的并且是最容易创建的实例之一。例如,在Scala中,我们可以简单地用(number1,number2)来创建一个含有两个数字的元组。除了元组之外,Scala的case类也是用来定义消息的理想类型——因为case类是不可变的、可以进行模式匹配并且还很容易进行复制。在Java中,我们可以通过将消息定义为一个不可修改(unmodifiable)的Collection的方式来将多个对象塞到一个消息中。当我们向角色传递消息时,如果发送者和接收者都在同一个JVM里[1],则默认情况下我们传递的是消息的引用。需要注意的是,保证所传递消息的不可变性是程序员自己的责任,尤其是当所发送的消息是我们自定义的类时则更需要加倍小心。为了解决这个问题,我们可以让Akka替我们先将消息序列化,然后将序列化出来的拷贝而不是原对象的引用发送出去,这样就可以避免由于类定义不严谨所造成的问题。

与角色交互最简单的方式莫过于“发送并忘记(fire and forget)”,即先将消息发出去,然后不等响应继续做下面的事。这种做法从性能角度考虑也是最好的选择。其中,发送动作是由调用方角色/线程发起的一个非阻塞操作。我们可以使用sendOneWay()函数或Scala的!函数来发送一个单向消息。

除了“发送并忘记”交互模式之外,Akka还提供了双向消息交互模式,以应对我们在发出消息之后需要等待对端角色响应的情况。在这种模式下,调用线程将被阻塞,直至收到对方响应或达到超时时间为止。下面让我们一起来看看如何在Java和Scala中收发消息。

在Java中收发消息

我们可以通过sendRequestReply()函数来发送消息并等待接收方响应。如果接收方的响应未在(可配置的)超时时间内到达,则系统将抛出ActorTimeoutException异常。下面让我们通过一个示例来学习这种双向消息通信方式:

01
public class FortuneTeller extends UntypedActor {
02
public void onReceive(final Object name) {
03
getContext().replyUnsafe(String.format("%s you'll rock", name));
04
}
05
public static void main(final String[] args) {
06
final ActorRef fortuneTeller =
07
Actors.actorOf(FortuneTeller.class).start();
08
try {
09
final Object response = fortuneTeller.sendRequestReply("Joe");
10
System.out.println(response);
11
} catch(ActorTimeoutException ex) {
12
System.out.println("Never got a response before timeout");
13
} finally {
14
fortuneTeller.stop();
15
}
16
}
17
}
在上面的代码中,我们定义了一个名为FortuneTeller的角色,它对收到的消息都会直接进行回复。为了发送回复消息,我们需要先调用getContext()函数获取调用上下文,然后再调用其replyUnsafe()函数来发送消息内容。调用replyUnsafe()函数所执行的发送动作是非阻塞的,并且请注意,在发送响应消息的过程中我们没有调用任何与角色有关的代码。在main()函数中我们调用了sendRequestReply()函数,该函数会在内部创建一个Future类并等待对方响应或超时抛出异常。下面让我们通过运行上述代码来看看Joe的命运如何:

Joe you’ll rock

我们上面实现的这个FortuneTeller实际上还有个问题没解决,即该角色依赖于消息发送者的发送方式。当消息发送方调用sendRequestReply()函数时,该函数会创建一个内部的Future用于等待对方响应。而如果我们换用sendOneWay()来发送消息的话,则replyUnsafe()函数将会失败。为了避免这种情况的发生,我们需要在调用replyUnsafe()函数之前先检查一下是否能匹配到一个处于阻塞状态的发送者。我们可以通过从上下文中读取发送者引用的方式来进行这个检查,也可以通过replySafe()函数的返回值来进行判断。因为当能取到发送者的引用时该函数会返回true,反之则返回false。下面我们就着手对FortuneTeller进行修改,以使其可以处理发送者没有阻塞地等待响应消息的情况:

01
public class FortuneTeller extends UntypedActor {
02
public void onReceive(final Object name) {
03
if(getContext().replySafe(String.format("%s you'll rock", name)))
04
System.out.println("Message sent for " + name);
05
else
06
System.out.println("Sender not found for " + name);
07
}
08
public static void main(final String[] args) {
09
final ActorRef fortuneTeller =
10
Actors.actorOf(FortuneTeller.class).start();
11
try {
12
fortuneTeller.sendOneWay("Bill");
13
final Object response = fortuneTeller.sendRequestReply("Joe");
14
System.out.println(response);
15
} catch(ActorTimeoutException ex) {
16
System.out.println("Never got a response before timeout");
17
} finally {
18
fortuneTeller.stop();
19
}
20
}
21
}
如上所示,新版的FortuneTeller代码很优雅地处理了我们之前提到的那些问题,即使接收方没找到发送者也不会导致处理失败。

1
Sender not found for Bill
2
Message sent for Joe
3
Joe you'll rock
我们知道,sendRequestReply()函数是需要等待对方响应的阻塞式函数,而sendOneWay()函数则是单向且非阻塞的。而如果既想要接收响应又不想被阻塞,则可以使用更复杂一些的sendRequestReplyFuture()函数。该函数可以返回一个Future对象,而拿到Future对象之后我们就可以继续干其他的事,直到我们真正需要用到对方的响应时,再选择阻塞式地等待或通过之前拿到的那个Future对象来查询对方的响应是否已经可用。类似地,在角色这一侧我们可以从上下文引用中取到senderFuture,并通过它来与发送方进行通信。在后面的示例中,我们会看到上述这些函数的具体用法,这里就不再赘述了。

请务必谨慎使用sendRequestReply()和sendRequestReplyFuture()函数,因为这两个函数都是阻塞的,所以调用它们对程序的性能和可扩展性都会造成负面影响。

在Scala中收发消息

如果想要在Scala中与角色进行消息收发,我们需要有些心理准备,因为在Scala中我们所采用的方法将会与Java API有不小的差别:

在Scala中,我们可以直接使用self属性来访问actor。通过该属性,我们可以调用reply()函数或replySafe()函数,其中reply()就是replyUnsafe()在Scala侧的等价函数。

在Scala中,我们既可以调用sendRequestReply()函数,也可以调用更优雅的 !! 函数——当然美是人者见仁智者见智的。同样地,!!!也可以被用来替换sendRequestReplyFuture()函数。

在Scala中,sendRequestReply()函数不再返回一个Object,而是返回一个Scala的Option。当接收方的响应抵达时,这个Option将是一个Some[T]的实例。该实例中存有响应的具体内容,而在超时的情况下则响应内容为None。所以,与Java版本所不同的是,在Scala中调用sendRequestReply()函数在超时的时候不会抛异常。

下面让我们先用不安全的reply()函数实现Scala版的FortuneTeller:

01
class FortuneTeller extends Actor {
02
def receive = {
03
case name : String =>
04
self.reply(String.format("%s you'll rock", name))
05
}
06
}
07
176 • Chapter 8. Favoring Isolated Mutability
08
object FortuneTeller {
09
def main(args : Array[String]) : Unit = {
10
val fortuneTeller = Actor.actorOf[FortuneTeller].start()
11
val response = fortuneTeller !! "Joe"
12
response match {
13
case Some(responseMessage) => println(responseMessage)
14
case None => println("Never got a response before timeout")
15
}
16
fortuneTeller.stop()
17
}
18
}
在角色的实现代码中,我们可以看到与Java版本的两点不同:其一是这里中我们用self代替了getContext()函数,另一个则是用reply()代替了replyUnsafe()函数。在调用方这一侧,我们使用了!!,即java中的sendRequestReply()函数来给角色发送消息,并在所收到的响应内容上应用了模式匹配。如果发送方收到对端的响应,则第一个case语句将被执行,而如果响应超时则第二个case语句将被执行。不出所料,该示例代码的运行结果与Java版完全相同:

1
Joe you'll rock
除了我们之前所讨论过的那些变更之外,安全版reply()函数的使用方式与Java版本差别不大。在Scala中,我们使用的是reply_?()或replySafe()。

01
class FortuneTeller extends Actor {
02
def receive = {
03
case name : String =>
04
if(self.reply_?(String.format("%s you'll rock", name)))
05
println("Message sent for " + name)
06
else
07
println("Sender not found for " + name)
08
}
09
}
10
object FortuneTeller {
11
def main(args : Array[String]) : Unit = {
12
val fortuneTeller = Actor.actorOf[FortuneTeller].start()
13
fortuneTeller ! "Bill"
14
val response = fortuneTeller !! "Joe"
15
response match {
16
case Some(responseMessage) => println(responseMessage)
17
case None => println("Never got a response before timeout")
18
}
19
fortuneTeller.stop()
20
}
21
}
通过上述修改,即使是在发送者未知的情况下,新版本的FortureTeller也不会失败:

1
Sender not found for Bill
2
Message sent for Joe
3
Joe you'll rock
Akka有一点很方便的行为就是,当我们用Akka发送消息时,它会将发送者的引用透明地传递过去。于是我们就无需显式地将发送者作为消息的一部分传递出去,从而省去了很多繁冗的代码。

如果不习惯使用像!、!!、!!!以及reply_?这样的函数名,我们也可以分别用sendOneWay()、sendRequestReply()、sendRequestReplyFuture()以及replySafe()这些函数来替换使用。

[1] Akka同样支持远程角色,以便使我们可以在不同机器的离散进程之间发送消息。

文章转自 并发编程网-ifeve.com

目录
相关文章
|
缓存 Ubuntu 搜索推荐
Ubuntu执行apt-get install xxx报错怎么办?
如果以上步骤无法解决问题,你可以在搜索引擎或Linux社区中查找更详细的解决方案。 买CN2云服务器,免备案服务器,高防服务器,就选蓝易云。百度搜索:蓝易云
1001 0
|
XML Java API
Apache POI详解及Word文档读取示例
apache poi资料详解,包括内部jar包依赖关系,及与使用文档的对应关系
4329 0
|
3天前
|
云安全 人工智能 运维
当攻击全面Agent化,你的安全防御能否跟上AI Speed?
多起客户侧AK泄露事件应急的真实观察
1214 3
|
26天前
|
人工智能 JSON API
全网刷屏的 Jev 模型正式开放!一手实战测评 + 保姆级教程
全网爆火的 Jev 模型是什么?有什么用?怎么使用?怎么接入 AI 编程工具?效果真的好么?傻子可懂的 Jev 保姆级实战教程 + 项目实战测评来啦
9642 32
|
12天前
|
存储 人工智能 并行计算
大模型本地部署终端选型方法论:以 Qwen3.8-27B 为例的四档分层完整流程
本文提出一套大模型本地部署终端选型方法论:定约束、定档位、定框架、定参数四步决策法,配合入门、主力、质量、无损四档分层模型。以 Qwen3.8-27B 实测数据为例,逐环节解读显存、带宽、存储、散热、系统、预算等要素,给出面向不同预算的优选方案、决策自查清单与市场观察框架。文末前瞻 AI 笔记本的 CPU+GPU 与统一内存两条路线,论证四步决策法在新品类上的延续性。
|
11天前
|
人工智能 JSON 自然语言处理
2026 年 Jev 决策模型深度拆解:原理解读、实战测评与保姆级落地教程
有一款特殊AI模型在开发者圈子刷屏,它摒弃传统大模型擅长的对话聊天能力,专注做高速结构化决策,它就是TypeSafe AI推出的Jev模型。该模型由ChatGPT共同发明人Diogo Almeida主导研发,定位为**System One Model(系统一模型)**,对标人类大脑快速直觉判断的思维模式,在响应延迟、调用成本、结构化输出稳定性上相比传统生成式大模型有着巨大差异。本文会完整拆解Jev底层原理、三大核心原语能力、适用业务场景,同时提供可直接运行的curl、Python代码示例,并且结合多组实测数据,客观分析模型优势与能力边界,帮助普通开发者和AI应用从业者快速上手落地。
824 1
|
7天前
|
人工智能 JSON 自然语言处理
2026 年 Jev 决策模型全解析:核心原理拆解,性能实战测评与落地部署教程
有一款特殊AI模型在开发者圈子刷屏,它摒弃传统大模型擅长的对话聊天能力,专注做高速结构化决策,它就是TypeSafe AI推出的Jev模型。该模型由ChatGPT共同发明人Diogo Almeida主导研发,定位为System One Model(系统一模型),对标人类大脑快速直觉判断的思维模式,在响应延迟、调用成本、结构化输出稳定性上相比传统生成式大模型有着巨大差异。本文会完整拆解Jev底层原理、三大核心原语能力、适用业务场景,同时提供可直接运行的curl、Python代码示例,并且结合多组实测数据,客观分析模型优势与能力边界,帮助普通开发者和AI应用从业者快速上手落地。
545 0