分布式模型的思想就是借鉴并发模型的基础上推演发展来的。

认识两个状态

并发模型的一个重要的方面是，线程是否应该共享状态，是具有共享状态还是独立状态。共享状态也就意味着在不同线程之间共享某些状态

状态其实就是数据，比如一个或者多个对象。当线程要共享数据时，就会造成竞态条件或者死锁等问题。当然，这些问题只是可能会出现，具体实现方式取决于你是否安全的使用和访问共享对象。

独立的状态表明状态不会在多个线程之间共享，如果线程之间需要通信的话，他们可以访问不可变的对象来实现，这是最有效的避免并发问题的一种方式，如下图所示

使用独立状态让我们的设计更加简单，因为只有一个线程能够访问对象，即使交换对象，也是不可变的对象。

并发模型

并行 Worker

第一个并发模型是并行 worker 模型，客户端会把任务交给代理人(Delegator)，然后由代理人把工作分配给不同的工人(worker)。如下图所示

并行 worker 的核心思想是，它主要

有两个进程即代理人和工人，Delegator 负责接收来自客户端的任务并把任务下发，交给具体的 Worker 进行处理，Worker 处理完成后把结果返回给 Delegator，在 Delegator 接收到 Worker 处理的结果后对其进行汇总，然后交给客户端。

并行 Worker 模型是 Java 并发模型中非常常见的一种模型。许多 java.util.concurrent 包下的并发工具都使用了这种模型。

并行 Worker 的优点

并行 Worker 模型的一个非常明显的特点就是很容易理解，为了提高系统的并行度你可以增加多个 Worker 完成任务。

并行 Worker 模型的另外一个好处就是，它会将一个任务拆分成多个小任务，并发执行，Delegator 在接受到 Worker 的处理结果后就会返回给 Client，整个 Worker -> Delegator -> Client 的过程是异步的。

并行 Worker 的缺点

同样的，并行 Worker 模式同样会有一些隐藏的缺点

共享状态会变得很复杂

实际的并行 Worker 要比我们图中画出的更复杂，主要是并行 Worker 通常会访问内存或共享数据库中的某些共享数据。

这些共享状态可能会使用一些工作队列来保存业务数据、数据缓存、数据库的连接池等。在线程通信中，线程需要确保共享状态是否能够让其他线程共享，而不是仅仅停留在 CPU 缓存中让自己可用，当然这些都是程序员在设计时就需要考虑的问题。线程需要避免竞态条件，死锁和许多其他共享状态造成的并发问题。

多线程在访问共享数据时，会丢失并发性，因为操作系统要保证只有一个线程能够访问数据，这会导致共享数据的争用和抢占。未抢占到资源的线程会阻塞。

现代的非阻塞并发算法可以减少争用提高性能，但是非阻塞算法比较难以实现。

可持久化的数据结构(Persistent data structures)是另外一个选择。可持久化的数据结构在修改后始终会保留先前版本。因此，如果多个线程同时修改一个可持久化的数据结构，并且一个线程对其进行了修改，则修改的线程会获得对新数据结构的引用。

虽然可持久化的数据结构是一个新的解决方法，但是这种方法实行起来却有一些问题，比如，一个持久列表会将新元素添加到列表的开头，并返回所添加的新元素的引用，但是其他线程仍然只持有列表中先前的第一个元素的引用，他们看不到新添加的元素。

持久化的数据结构比如链表(LinkedList)在硬件性能上表现不佳。列表中的每个元素都是一个对象，这些对象散布在计算机内存中。现代 CPU 的顺序访问往往要快的多，因此使用数组等顺序访问的数据结构则能够获得更高的性能。CPU 高速缓存可以将一个大的矩阵块加载到高速缓存中，并让 CPU 在加载后直接访问 CPU 高速缓存中的数据。对于链表，将元素分散在整个 RAM 上，这实际上是不可能的。

无状态的 worker

共享状态可以由其他线程所修改，因此，worker 必须在每次操作共享状态时重新读取，以确保在副本上能够正确工作。不在线程内部保持状态的 worker 成为无状态的 worker。

作业顺序是不确定的

并行工作模型的另一个缺点是作业的顺序不确定，无法保证首先执行或最后执行哪些作业。任务 A 在任务 B 之前分配给 worker，但是任务 B 可能在任务 A 之前执行。

流水线

第二种并发模型就是我们经常在生产车间遇到的流水线并发模型，下面是流水线设计模型的流程图

这种组织架构就像是工厂中装配线中的 worker，每个 worker 只完成全部工作的一部分，完成一部分后，worker 会将工作转发给下一个 worker。

每道程序都在自己的线程中运行，彼此之间不会共享状态，这种模型也被称为无共享并发模型。

使用流水线并发模型通常被设计为非阻塞 I/O，也就是说，当没有给 worker 分配任务时，worker 会做其他工作。非阻塞 I/O 意味着当 worker 开始 I/O 操作，例如从网络中读取文件，worker 不会等待 I/O 调用完成。因为 I/O 操作很慢，所以等待 I/O 非常耗费时间。在等待 I/O 的同时，CPU 可以做其他事情，I/O 操作完成后的结果将传递给下一个 worker。下面是非阻塞 I/O 的流程图

在实际情况中，任务通常不会按着一条装配线流动，由于大多数程序需要做很多事情，因此需要根据完成的不同工作在不同的 worker 之间流动，如下图所示

任务还可能需要多个 worker 共同参与完成

响应式 – 事件驱动系统

使用流水线模型的系统有时也被称为响应式或者事件驱动系统，这种模型会根据外部的事件作出响应，事件可能是某个 HTTP 请求或者某个文件完成加载到内存中。

Actor 模型

在 Actor 模型中，每一个 Actor 其实就是一个 Worker， 每一个 Actor 都能够处理任务。

简单来说，Actor 模型是一个并发模型，它定义了一系列系统组件应该如何动作和交互的通用规则，最著名的使用这套规则的编程语言是 Erlang。一个参与者 Actor 对接收到的消息做出响应，然后可以创建出更多的 Actor 或发送更多的消息，同时准备接收下一条消息。

Channels 模型

在 Channel 模型中，worker 通常不会直接通信，与此相对的，他们通常将事件发送到不同的通道(Channel)上，然后其他 worker 可以在这些通道上获取消息，下面是 Channel 的模型图

有的时候 worker 不需要明确知道接下来的 worker 是谁，他们只需要将作者写入通道中，监听 Channel 的 worker 可以订阅或者取消订阅，这种方式降低了 worker 和 worker 之间的耦合性。

流水线设计的优点

与并行设计模型相比，流水线模型具有一些优势，具体优势如下

不会存在共享状态

因为流水线设计能够保证 worker 在处理完成后再传递给下一个 worker，所以 worker 与 worker 之间不需要共享任何状态，也就无需考虑并发问题。你甚至可以在实现上把每个 worker 看成是单线程的一种。

有状态 worker

因为 worker 知道没有其他线程修改自身的数据，所以流水线设计中的 worker 是有状态的，有状态的意思是他们可以将需要操作的数据保留在内存中，有状态通常比无状态更快。

更好的硬件整合

因为你可以把流水线看成是单线程的，而单线程的工作优势在于它能够和硬件的工作方式相同。因为有状态的 worker 通常在 CPU 中缓存数据，这样可以更快地访问缓存的数据。

使任务更加有效的进行

可以对流水线并发模型中的任务进行排序，一般用来日志的写入和恢复。

流水线设计的缺点

流水线并发模型的缺点是任务会涉及多个 worker，因此可能会分散在项目代码的多个类中。因此很难确定每个 worker 都在执行哪个任务。流水线的代码编写也比较困难，设计许多嵌套回调处理程序的代码通常被称为回调地狱。回调地狱很难追踪 debug。

函数性并行

函数性并行模型是最近才提出的一种并发模型，它的基本思路是使用函数调用来实现。消息的传递就相当于是函数的调用。传递给函数的参数都会被拷贝，因此在函数之外的任何实体都无法操纵函数内的数据。这使得函数执行类似于原子操作。每个函数调用都可以独立于任何其他函数调用执行。

当每个函数调用独立执行时，每个函数都可以在单独的 CPU 上执行。这也就是说，函数式并行并行相当于是各个 CPU 单独执行各自的任务。

JDK 1.7 中的 ForkAndJoinPool 类就实现了函数性并行的功能。Java 8 提出了 stream 的概念，使用并行流也能够实现大量集合的迭代。

函数性并行的难点是要知道函数的调用流程以及哪些 CPU 执行了哪些函数，跨 CPU 函数调用会带来额外的开销。

我们之前说过，线程就是进程中的一条顺序流，在 Java 中，每一条 Java 线程就像是 JVM 的一条顺序流，就像是虚拟 CPU 一样来执行代码。Java 中的 main()方法是一条特殊的线程，JVM 创建的 main 线程是一条主执行线程，在 Java 中，方法都是由 main 方法发起的。在 main 方法中，你照样可以创建其他的线程(执行顺序流)，这些线程可以和 main 方法共同执行应用代码。

Java 线程也是一种对象，它和其他对象一样。Java 中的 Thread 表示线程，Thread 是 java.lang.Thread 类或其子类的实例。那么下面我们就来一起探讨一下在 Java 中如何创建和启动线程。

创建并启动线程

在 Java 中，创建线程的方式主要有三种

• 通过继承 Thread 类来创建线程
• 通过实现 Runnable 接口来创建线程
• 通过 Callable 和 Future 来创建线程

下面我们分别探讨一下这几种创建方式

继承 Thread 类来创建线程

第一种方式是继承 Thread 类来创建线程，如下示例

线程的主要创建步骤如下

• 定义一个线程类使其继承 Thread 类，并重写其中的 run 方法，run 方法内部就是线程要完成的任务，因此 run 方法也被称为执行体
• 创建了 Thread 的子类，上面代码中的子类是 TJavaThread
• 启动方法需要注意，并不是直接调用 run 方法来启动线程，而是使用 start 方法来启动线程。当然 run 方法可以调用，这样的话就会变成普通方法调用，而不是新创建一个线程来调用了。

这样的话，整个 main 方法只有一条执行线程也就是 main 线程，由两条执行线程变为一条执行线程

Thread 构造器只需要一个 Runnable 对象，调用 Thread 对象的 start() 方法为该线程执行必须的初始化操作，然后调用 Runnable 的 run 方法，以便在这个线程中启动任务。我们上面使用了线程的 join 方法，它用来等待线程的执行结束，如果我们不加 join 方法，它就不会等待 tJavaThread 的执行完毕，输出的结果可能就不是 10000

可以看到，在 run 方法还没有结束前，run 就被返回了。也就是说，程序不会等到 run 方法执行完毕就会执行下面的指令。

使用继承方式创建线程的优势：编写比较简单；可以使用 this 关键字直接指向当前线程，而无需使用 Thread.currentThread()来获取当前线程。

使用继承方式创建线程的劣势：在 Java 中，只允许单继承（拒绝肛精说使用内部类可以实现多继承）的原则，所以使用继承的方式，子类就不能再继承其他类。

使用 Runnable 接口来创建线程

相对的，还可以使用 Runnable 接口来创建线程，如下示例

线程的主要创建步骤如下

• 首先定义 Runnable 接口，并重写 Runnable 接口的 run 方法，run 方法的方法体同样是该线程的线程执行体。
• 创建线程实例，可以使用上面代码这种简单的方式创建，也可以通过 new 出线程的实例来创建，如下所示

• 再调用线程对象的 start 方法来启动该线程。

线程在使用实现 Runnable 的同时也能实现其他接口，非常适合多个相同线程来处理同一份资源的情况，体现了面向对象的思想。

使用 Runnable 实现的劣势是编程稍微繁琐，如果要访问当前线程，则必须使用 Thread.currentThread()方法。

使用 Callable 接口来创建线程

Runnable 接口执行的是独立的任务，Runnable 接口不会产生任何返回值，如果你希望在任务完成后能够返回一个值的话，那么你可以实现 Callable 接口而不是 Runnable 接口。Java SE5 引入了 Callable 接口，它的示例如下

我想，使用 Callable 接口的好处你已经知道了吧，既能够实现多个接口，也能够得到执行结果的返回值。Callable 和 Runnable 接口还是有一些区别的，主要区别如下

• Callable 执行的任务有返回值，而 Runnable 执行的任务没有返回值
• Callable（重写）的方法是 call 方法，而 Runnable（重写）的方法是 run 方法。
• call 方法可以抛出异常，而 Runnable 方法不能抛出异常

使用线程池来创建线程

首先先来认识一下顶级接口 Executor，Executor 虽然不是传统线程创建的方式之一，但是它却成为了创建线程的替代者，使用线程池的好处如下

• 利用线程池能够复用线程、控制最大并发数。
• 实现任务线程队列缓存策略和拒绝机制。
• 实现某些与时间相关的功能，如定时执行、周期执行等。
• 隔离线程环境。比如，交易服务和搜索服务在同一台服务器上，分别开启两个线程池，交易线程的资源消耗明显要大；因此，通过配置独立的线程池，将较慢的交易服务与搜索服务隔开，避免个服务线程互相影响。

你可以使用如下操作来替换线程创建

new Thread(new(RunnableTask())).start()

// 替换为

Executor executor = new ExecutorSubClass() // 线程池实现类;executor.execute(new RunnableTask1());executor.execute(new RunnableTask2());

ExecutorService 是 Executor 的默认实现，也是 Executor 的扩展接口，ThreadPoolExecutor 类提供了线程池的扩展实现。Executors 类为这些 Executor 提供了方便的工厂方法。下面是使用 ExecutorService 创建线程的几种方式

CachedThreadPool

从而简化了并发编程。Executor 在客户端和任务之间提供了一个间接层；与客户端直接执行任务不同，这个中介对象将执行任务。Executor 允许你管理异步任务的执行，而无须显示地管理线程的生命周期。

public static void main(String[] args) {ExecutorService service = Executors.newCachedThreadPool();for(int i = 0;i < 5;i++){service.execute(new TestThread());}service.shutdown();}

CachedThreadPool 会为每个任务都创建一个线程。

注意：ExecutorService 对象是使用静态的 Executors 创建的，这个方法可以确定 Executor 类型。对 shutDown 的调用可以防止新任务提交给 ExecutorService ，这个线程在 Executor 中所有任务完成后退出。

FixedThreadPool

FixedThreadPool 使你可以使用有限的线程集来启动多线程

public static void main(String[] args) {ExecutorService service = Executors.newFixedThreadPool(5);for(int i = 0;i < 5;i++){service.execute(new TestThread());}service.shutdown();}

有了 FixedThreadPool 使你可以一次性的预先执行高昂的线程分配，因此也就可以限制线程的数量。这可以节省时间，因为你不必为每个任务都固定的付出创建线程的开销。

SingleThreadExecutor

SingleThreadExecutor 就是线程数量为 1 的 FixedThreadPool，如果向 SingleThreadPool 一次性提交了多个任务，那么这些任务将会排队，每个任务都会在下一个任务开始前结束，所有的任务都将使用相同的线程。SingleThreadPool 会序列化所有提交给他的任务，并会维护它自己(隐藏)的悬挂队列。

public static void main(String[] args) {ExecutorService service = Executors.newSingleThreadExecutor();for(int i = 0;i < 5;i++){service.execute(new TestThread());}service.shutdown();}

从输出的结果就可以看到，任务都是挨着执行的。我为任务分配了五个线程，但是这五个线程不像是我们之前看到的有换进换出的效果，它每次都会先执行完自己的那个线程，然后余下的线程继续走完这条线程的执行路径。你可以用 SingleThreadExecutor 来确保任意时刻都只有唯一一个任务在运行。

休眠

影响任务行为的一种简单方式就是使线程 休眠，选定给定的休眠时间，调用它的 sleep()方法， 一般使用的 TimeUnit 这个时间类替换 Thread.sleep()方法，示例如下：

public class SuperclassThread extends TestThread{

@Overridepublic void run() {System.out.println(Thread.currentThread() + “starting …” );

try {for(int i = 0;i < 5;i++){if(i == 3){System.out.println(Thread.currentThread() + "sleeping …");TimeUnit.MILLISECONDS.sleep(1000);}}} catch (InterruptedException e) {e.printStackTrace();}

System.out.println(Thread.currentThread() + “wakeup and end …”);}

public static void main(String[] args) {ExecutorService executors = Executors.newCachedThreadPool();for(int i = 0;i < 5;i++){executors.execute(new SuperclassThread());}executors.shutdown();}}

关于 TimeUnit 中的 sleep() 方法和 Thread.sleep() 方法的比较，请参考下面这篇博客

([www.cnblogs.com/xiadongqing…](

))

优先级

上面提到线程调度器对每个线程的执行都是不可预知的，随机执行的，那么有没有办法告诉线程调度器哪个任务想要优先被执行呢？你可以通过设置线程的优先级状态，告诉线程调度器哪个线程的执行优先级比较高，请给这个骑手马上派单，线程调度器倾向于让优先级较高的线程优先执行，然而，这并不意味着优先级低的线程得不到执行，也就是说，优先级不会导致死锁的问题。优先级较低的线程只是执行频率较低。

public class SimplePriorities implements Runnable{

private int priority;

public SimplePriorities(int priority) {this.priority = priority;}

@Overridepublic void run() {Thread.currentThread().setPriority(priority);for(int i = 0;i < 100;i++){System.out.println(this);if(i % 10 == 0){Thread.yield();}}}

@Overridepublic String toString() {return Thread.currentThread() + ” ” + priority;}

public static void main(String[] args) {ExecutorService service = Executors.newCachedThreadPool();for(int i = 0;i < 5;i++){service.execute(new SimplePriorities(Thread.MAX_PRIORITY));}service.execute(new SimplePriorities(Thread.MIN_PRIORITY));}}

toString() 方法被覆盖，以便通过使用 Thread.toString()方法来打印线程的名称。你可以改写线程的默认输出，这里采用了 Thread[pool-1-thread-1,10,main] 这种形式的输出。

通过输出，你可以看到，最后一个线程的优先级最低，其余的线程优先级最高。注意，优先级是在 run 开头设置的，在构造器中设置它们不会有任何好处，因为这个时候线程还没有执行任务。

尽管 JDK 有 10 个优先级，但是一般只有 MAX_PRIORITY，NORM_PRIORITY，MIN_PRIORITY 三种级别。

作出让步

我们上面提过，如果知道一个线程已经在 run() 方法中运行的差不多了，那么它就可以给线程调度器一个提示：我已经完成了任务中最重要的部分，可以让给别的线程使用 CPU 了。这个暗示将通过 yield() 方法作出。

有一个很重要的点就是，Thread.yield() 是建议执行切换 CPU，而不是强制执行 CPU 切换。

对于任何重要的控制或者在调用应用时，都不能依赖于 yield()方法，实际上， yield() 方法经常被滥用。

后台线程

后台(daemon)线程，是指运行时在后台提供的一种服务线程，这种线程不是属于必须的。当所有非后台线程结束时，程序也就停止了，**同时会终止所有的后台线程。**反过来说，只要有任何非后台线程还在运行，程序就不会终止。

public class SimpleDaemons implements Runnable{

@Overridepublic void run() {while (true){try {TimeUnit.MILLISECONDS.sleep(100);System.out.println(Thread.currentThread() + ” ” + this);} catch (InterruptedException e) {System.out.println(“sleep() interrupted”);}}}

public static void main(String[] args) throws InterruptedException {for(int i = 0;i < 10;i++){Thread daemon = new Thread(new SimpleDaemons());daemon.setDaemon(true);daemon.start();}System.out.println("All Daemons started");TimeUnit.MILLISECONDS.sleep(175);}}

在每次的循环中会创建 10 个线程，并把每个线程设置为后台线程，然后开始运行，for 循环会进行十次，然后输出信息，随后主线程睡眠一段时间后停止运行。在每次 run 循环中，都会打印当前线程的信息，主线程运行完毕，程序就执行完毕了。因为 daemon 是后台线程，无法影响主线程的执行。

但是当你把 daemon.setDaemon(true)去掉时，while(true) 会进行无限循环，那么主线程一直在执行最重要的任务，所以会一直循环下去无法停止。

ThreadFactory

按需要创建线程的对象。使用线程工厂替换了 Thread 或者 Runnable 接口的硬连接，使程序能够使用特殊的线程子类，优先级等。一般的创建方式为

class SimpleThreadFactory implements ThreadFactory {public Thread newThread(Runnable r) {return new Thread(r);}}

Executors.defaultThreadFactory 方法提供了一个更有用的简单实现，它在返回之前将创建的线程上下文设置为已知值

ThreadFactory 是一个接口，它只有一个方法就是创建线程的方法

public interface ThreadFactory {

// 构建一个新的线程。实现类可能初始化优先级，名称，后台线程状态和 线程组等 Thread newThread(Runnable r);}

下面来看一个 ThreadFactory 的例子

public class DaemonThreadFactory implements ThreadFactory {

@Overridepublic Thread newThread(Runnable r) {Thread t = new Thread(r);t.setDaemon(true);return t;}}

public class DaemonFromFactory implements Runnable{

@Overridepublic void run() {while (true){try {TimeUnit.MILLISECONDS.sleep(100);System.out.println(Thread.currentThread() + ” ” + this);} catch (InterruptedException e) {System.out.println(“Interrupted”);}}}

public static void main(String[] args) throws InterruptedException {ExecutorService service = Executors.newCachedThreadPool(new DaemonThreadFactory());for(int i = 0;i < 10;i++){service.execute(new DaemonFromFactory());}System.out.println("All daemons started");TimeUnit.MILLISECONDS.sleep(500);}}

Executors.newCachedThreadPool 可以接受一个线程池对象，创建一个根据需要创建新线程的线程池，但会在它们可用时重用先前构造的线程，并在需要时使用提供的 ThreadFactory 创建新线程。

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue(),threadFactory);}

加入一个线程

一个线程可以在其他线程上调用 join()方法，其效果是等待一段时间直到第二个线程结束才正常执行。如果某个线程在另一个线程 t 上调用 t.join() 方法，此线程将被挂起，直到目标线程 t 结束才回复(可以用 t.isAlive() 返回为真假判断)。

也可以在调用 join 时带上一个超时参数，来设置到期时间，时间到期，join 方法自动返回。

对 join 的调用也可以被中断，做法是在线程上调用 interrupted 方法，这时需要用到 try…catch 子句

public class TestJoinMethod extends Thread{

@Overridepublic void run() {for(int i = 0;i < 5;i++){try {TimeUnit.MILLISECONDS.sleep(1000);} catch (InterruptedException e) {System.out.println("Interrupted sleep");}System.out.println(Thread.currentThread() + " " + i);}}

public static void main(String[] args) throws InterruptedException {TestJoinMethod join1 = new TestJoinMethod();TestJoinMethod join2 = new TestJoinMethod();TestJoinMethod join3 = new TestJoinMethod();

join1.start();// join1.join();

join2.start();join3.start();}}

join() 方法等待线程死亡。 换句话说，它会导致当前运行的线程停止执行，直到它加入的线程完成其任务。

线程异常捕获

由于线程的本质，使你不能捕获从线程中逃逸的异常，一旦异常逃出任务的 run 方法，它就会向外传播到控制台，除非你采取特殊的步骤捕获这种错误的异常，在 Java5 之前，你可以通过线程组来捕获，但是在 Java 5 之后，就需要用 Executor 来解决问题，因为线程组不是一次好的尝试。

下面的任务会在 run 方法的执行期间抛出一个异常，并且这个异常会抛到 run 方法的外面，而且 main 方法无法对它进行捕获

public class ExceptionThread implements Runnable{

@Overridepublic void run() {throw new RuntimeException();}

public static void main(String[] args) {try {ExecutorService service = Executors.newCachedThreadPool();service.execute(new ExceptionThread());

小伙伴们有兴趣想了解内容和更多相关学习资料的请点赞收藏+评论转发+关注我，后面会有很多干货。我有一些面试题、架构、设计类资料可以说是程序员面试必备！所有资料都整理到网盘了，需要的话欢迎下载！私信我回复【666】即可免费获取

作者：Java架构追梦

原文出处：https://xie.infoq.cn/article/ba889a8db90d444d24c67ffbc