布鲁
2023/03/19阅读:23主题:科技蓝
懒人专用高并发:Actor模型
传统多线程实现方式
public class MultiThreadExample implements Runnable {
private String threadName;
public MultiThreadExample(String name) {
this.threadName = name;
}
public void run() {
System.out.println("Thread " + threadName + " starting.");
for (int i = 0; i < 5; i++) {
System.out.println("Thread " + threadName + " running. Count: " + i);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
System.out.println("Thread " + threadName + " interrupted.");
}
}
System.out.println("Thread " + threadName + " exiting.");
}
public static void main(String[] args) {
System.out.println("Main thread starting.");
MultiThreadExample thread1 = new MultiThreadExample("Thread 1");
MultiThreadExample thread2 = new MultiThreadExample("Thread 2");
Thread t1 = new Thread(thread1);
Thread t2 = new Thread(thread2);
t1.start();
t2.start();
System.out.println("Main thread exiting.");
}
}
在这个例子中,我们定义了一个实现Runnable接口的类,这个接口定义了一个run()方法,用于执行线程的代码。
我们创建了两个线程实例,并使用Thread类将它们包装起来。我们调用start()方法来启动线程,然后每个线程都会打印5条消息,每条消息之间间隔1秒。当run()方法完成后,线程退出并打印一条退出消息。
基于Actor模型的实现方式
import akka.actor.*;
import java.util.concurrent.*;
public class AkkaExample extends UntypedActor {
private String threadName;
public AkkaExample(String name) {
this.threadName = name;
}
@Override
public void onReceive(Object message) throws Throwable {
if (message instanceof String) {
String msg = (String) message;
if (msg.equals("start")) {
System.out.println("Thread " + threadName + " starting.");
for (int i = 0; i < 5; i++) {
System.out.println("Thread " + threadName + " running. Count: " + i);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
System.out.println("Thread " + threadName + " interrupted.");
}
}
System.out.println("Thread " + threadName + " exiting.");
}
} else {
unhandled(message);
}
}
public static void main(String[] args) throws Exception {
System.out.println("Main thread starting.");
ActorSystem system = ActorSystem.create("mySystem");
ActorRef thread1 = system.actorOf(Props.create(AkkaExample.class, "Thread 1"));
ActorRef thread2 = system.actorOf(Props.create(AkkaExample.class, "Thread 2"));
thread1.tell("start", null);
thread2.tell("start", null);
system.terminate();
System.out.println("Main thread exiting.");
}
}
在这个例子中,我们定义了一个继承UntypedActor类的类,重写了onReceive()方法,实现了线程的执行逻辑。
我们创建了两个Actor实例,并使用ActorRef的tell()方法来向Actor发送消息启动它们。当系统终止时,所有Actor都已完成。
优势在哪里?
从这个例子中,可能看不出Akka框架对比传统多线程的优势。
下面是一个更能体现Akka框架优势的例子:
假设我们有一个需要并行处理的任务,该任务需要访问外部资源,可能是数据库、网络等等。传统多线程的做法是使用synchronized或ReentrantLock等锁机制来保证资源访问的安全性,但这会带来锁竞争、死锁等问题。
而Akka框架则采用了Actor模型,每个Actor都是独立的,拥有自己的状态和消息队列,从而避免了锁的问题。
import akka.actor.*;
import java.util.concurrent.*;
public class ExternalResourceExample {
public static void main(String[] args) throws Exception {
System.out.println("Main thread starting.");
ActorSystem system = ActorSystem.create("mySystem");
ActorRef requester = system.actorOf(Props.create(Requester.class));
requester.tell("start", null);
system.terminate();
System.out.println("Main thread exiting.");
}
public static class Requester extends UntypedActor {
private final ActorRef worker1;
private final ActorRef worker2;
private final CountDownLatch latch;
private long startTime;
public Requester() {
this.worker1 = getContext().actorOf(Props.create(Worker.class));
this.worker2 = getContext().actorOf(Props.create(Worker.class));
this.latch = new CountDownLatch(2);
this.startTime = 0;
}
@Override
public void onReceive(Object message) throws Throwable {
if (message instanceof String) {
String msg = (String) message;
if (msg.equals("start")) {
startTime = System.currentTimeMillis();
System.out.println("Requester starting.");
worker1.tell("request", getSelf());
worker2.tell("request", getSelf());
} else if (msg.equals("response")) {
latch.countDown();
if (latch.getCount() == 0) {
long endTime = System.currentTimeMillis();
System.out.println("Requester exiting. Total time: " + (endTime - startTime) + "ms");
getContext().system().terminate();
}
}
} else {
unhandled(message);
}
}
}
public static class Worker extends UntypedActor {
private final Random random;
public Worker() {
this.random = new Random();
}
@Override
public void onReceive(Object message) throws Throwable {
if (message instanceof String) {
String msg = (String) message;
if (msg.equals("request")) {
System.out.println("Worker " + getSelf().path().name() + " starting.");
long sleepTime = random.nextInt(5000);
try {
Thread.sleep(sleepTime);
} catch (InterruptedException e) {
System.out.println("Worker " + getSelf().path().name() + " interrupted.");
}
System.out.println("Worker " + getSelf().path().name() + " done. Time: " + sleepTime + "ms");
getSender().tell("response", getSelf());
}
} else {
unhandled(message);
}
}
}
}
在这个例子中,我们定义了两个Actor,一个Requester和一个Worker。
Requester会向两个Worker发送消息,请求它们模拟请求外部资源的操作。每个Worker会随机休眠一段时间(模拟请求耗时),然后向Requester发送响应消息。Requester在收到两个Worker的响应后退出。
我们来看一下这个例子相对于传统多线程的优势:
-
代码更加简洁易读。相比传统多线程,使用Akka框架的代码更加简洁易读。使用Actor模型,我们只需要定义每个Actor的行为和如何相互发送消息,而不需要显式地使用锁机制来保证线程安全性。 -
更好的可伸缩性。在传统多线程中,使用锁机制来保证线程安全性往往会导致锁竞争,从而影响性能和可伸缩性。而在Akka框架中,每个Actor都是独立的,相互之间没有锁竞争问题,从而更加容易实现可伸缩性。 -
更好的容错性。在传统多线程中,由于使用锁机制来保证线程安全性,一旦出现死锁或其他问题,很难进行容错处理。而在Akka框架中,由于每个Actor是独立的,相互之间没有锁竞争问题,因此更容易实现容错处理。例如,当一个Actor出现故障时,可以很容易地停止该Actor并创建一个新的Actor来替代它,而不会影响其他Actor的正常运行。
背景知识
Akka框架
Akka是一个开源的、基于Actor模型的并发编程框架。Actor模型是一种轻量级的并发编程模型,它将并发程序看作是一组相互独立的、互相发送消息的Actor(类似于面向对象编程中的对象),每个Actor都有自己的状态和行为,它们之间通过异步消息传递进行通信。
Akka框架提供了Actor模型的实现,同时还提供了一些工具和库,帮助开发人员构建高并发、分布式、可伸缩的应用程序。Akka框架具有以下特点:
-
高性能:Akka框架使用异步非阻塞的I/O模型,采用事件驱动的方式处理消息,从而提高了并发性能。
-
可伸缩性:Akka框架中每个Actor都是独立的,相互之间没有共享状态,从而避免了锁竞争等问题,使得应用程序可以方便地进行横向扩展。
-
容错性:Akka框架提供了内置的容错机制,当某个Actor发生故障时,Akka框架可以自动地重新启动该Actor,并将其状态恢复到故障发生之前。
-
分布式:Akka框架支持分布式部署,可以将不同的Actor部署到不同的机器上,从而实现高可用性和容错性。
-
易于使用:Akka框架使用Java和Scala两种语言进行开发,提供了简洁易用的API,方便开发人员进行编程。
Actor模型

Actor模型是一种并行计算模型,提供了一种用于构建并发、分布式系统的抽象方法。
在Actor模型中,计算被表示为独立的、轻量级的计算单元,称为Actor,它们可以发送和接收消息并进行本地计算。
Actor模型中的每个Actor都是一个独立的运行实体,具有自己的状态和行为。
Actor之间通过消息传递进行通信,每个Actor都有一个邮箱(Mailbox)来接收消息,每次只处理一个消息,以确保状态的一致性和线程安全。
Actor模型中的消息传递是异步的,发送方不需要等待接收方的响应,从而避免了锁和同步的开销。
同时,Actor之间是完全独立的,不存在共享内存和竞争条件的问题,可以避免并发编程中的一些难点。
Actor模型的优点是能够提供高度的并发性和可扩展性,能够有效地解决多核CPU和分布式系统中的并发编程问题。
它还能够提供良好的容错性和可恢复性,因为每个Actor都有自己的状态和行为,可以更容易地实现系统的容错和恢复。
Actor模型已经在许多编程语言中得到了实现,比如Erlang、Akka(基于Java语言实现)、Scala等。
Fork/Join框架
Fork/Join框架是Java 7中提供的一个并行计算框架,它是基于“分治法”思想实现的。该框架的目标是提高多核CPU下的计算性能,通过利用多线程同时处理子任务,从而缩短程序的执行时间。
Fork/Join框架的主要特点如下:
-
分治思想:该框架基于“分治法”思想,将一个大的任务拆分成若干个小的任务,然后分别处理,并将结果合并。
-
工作窃取算法:该框架采用工作窃取算法,当某个线程的任务执行完毕后,可以从其他线程的任务队列中窃取任务继续执行,从而提高了任务的并行度。
-
高效性:该框架通过合理利用多线程,减少了线程的创建和销毁的开销,并且避免了线程之间的竞争和锁等问题,从而提高了程序的性能。
-
易于使用:该框架提供了简洁的API,开发人员可以很容易地将自己的任务拆分成小的子任务,并使用该框架来实现多线程并行计算。
Fork/Join框架的应用场景包括并行计算密集型任务,如图像处理、数字信号处理、科学计算等。它适用于能够将一个大的任务拆分成小的子任务,并可以并行处理的场景。
作者介绍