懒人专用高并发:Actor模型
创始人
2025-05-30 12:58:16
0

传统多线程实现方式

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;}@Overridepublic 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;}@Overridepublic 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();}@Overridepublic 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的响应后退出。

我们来看一下这个例子相对于传统多线程的优势:

  1. 代码更加简洁易读。相比传统多线程,使用Akka框架的代码更加简洁易读。使用Actor模型,我们只需要定义每个Actor的行为如何相互发送消息,而不需要显式地使用锁机制来保证线程安全性。
  2. 更好的可伸缩性。在传统多线程中,使用锁机制来保证线程安全性往往会导致锁竞争,从而影响性能和可伸缩性。而在Akka框架中,每个Actor都是独立的,相互之间没有锁竞争问题,从而更加容易实现可伸缩性。
  3. 更好的容错性。在传统多线程中,由于使用锁机制来保证线程安全性,一旦出现死锁或其他问题,很难进行容错处理。而在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框架的应用场景包括并行计算密集型任务,如图像处理、数字信号处理、科学计算等。它适用于能够将一个大的任务拆分成小的子任务,并可以并行处理的场景。

相关内容

热门资讯

块级元素和行内元素 一、块级元素和行内元素 行内元素:         常见的行内元素: a、span、b、img、st...
shell 部分内置变量 说明 这些变量都是由bash这个程序本身进行修改使用; 其他shell不一定试用; 变量 $_ to...
链表(没做完。。) BM1 反转链表 给定一个单链表的头结点pHead(该头节点是有值的,比如在下图&#...
十九、互斥量(互斥锁) 文章目录1、基本概念2、使用场景:用于实现对临界资源的独占式处理(能够解...
springboot控制层消息... 大概了解三种方式: 1.区别在于WebMvcConfigurerAdapter 新版本...
2023年湖北安全员ABC证书...   2023年湖北安全员ABC证书报名入口!报考条件?启程别 证...
蓝桥杯倒计时 | 倒计时18天 作者🕵️‍♂️:让机器理解语言か 专栏🎇:...
pandas读CSV、读JSO... 学习让我快乐   pandas的数据读取基本操作 pandas是Python中非常流行的数据处理库...
Hybrid App开发模式 Hybrid App(混合模式移动应用)是指介于web-app、nati...
SQL预编译和批量提交对性能提... 背景 一个项目,从kafka获取数据后,经过业务处理,生成...
C++ Lambda表达式的常... ⭐️我叫忆_恒心,一名喜欢书写博客的在读研究生👨‍🎓。...
【总结】docker 安装教程 各操作系统版本下载地址 mac 版:https://download.docker.c...
12 致远OA开发规范 开发案例(业务扩展接口) 1.客开开发前缀约束: 为了快速区分标准开发与客开开发&#x...
文件包含漏洞全面详解 文件包含漏洞总结一、什么是文件包含漏洞二、本地文件包含漏洞(LFI)三、LFI漏洞利用技巧1.配合文...
【C++进阶】unordere... 文章目录unordered系列容器介绍unordered_setunordered_set的模板参数...
【Linux】学会这些基本指令... 前言上篇文章介绍了一些常用的指令,这篇文章再来介绍一下Linux必须学会的指令。一.时...
panbas学习篇(一)数据的... 最大值最小值 Series.sum : Return the sum.Series.min : Re...
西安石油大学期末C语言重点题目... C语言重点知识点总结 ##例题4-7 (switch和字符的读入) ...
QML- QML视觉元素类型 QML视觉元素类型一、概述一、图像类型三、共享视觉属性1. 不透明度和可见性2. 转换(...
【 Linux入门 】之 手搓... 目的基本结构提取输入命令fgets的使用命令初步处理命令的本质创建子进程重要知识补充进程替换命令处理...