墨滴

布鲁

V1

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的响应后退出。

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

  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框架的应用场景包括并行计算密集型任务，如图像处理、数字信号处理、科学计算等。它适用于能够将一个大的任务拆分成小的子任务，并可以并行处理的场景。

分类：

后端

标签：

Java

作者介绍

布鲁
V1