Java并发编程基础03-高级线程同步机制

简介

3种基本的同步机制:

  • synchronized关键字。
  • Lock接口及其实现类:ReentrantLock、ReentrantReadWriteLock.ReadLock和ReentrantReadWriteLock.writeLock。
  • StampedLock类。

高级同步机制:

  • Semaphore:这是一个计数器,用来控制一个或多个共享资源的访问。这是一种基本并发编程工具,为大多数编程语言所支持。
  • CountDownLatch:CountDownLatch类是Java语言提供的一个机制,可以使一个线程等待多个操作结束。
  • CyclicBarrier:CyclicBarrier类是Java语言提供的另一个机制,可以使多个线程在一个共同状态点同步。
  • Phaser:Phaser类也是Java语言提供的另一个机制,可以分阶段地控制并发任务的执行。只有所有的线程都完成一个阶段后,才能继续下一个阶段。
  • Exchanger:Exchanger类也是Java语言提供的另一个机制,能够使得两个线程在某一点进行数据交换。
  • CompletableFuture:CompletableFuture类提供了一个机制,在这个机制中一个或多个任务等待另外一些任务执行结束,并且这些任务将以异步方式运行。这个类在Java 8中引入,Java 9中新增了一些方法。

信号量机制

如果信号量内部的计数器的值大于0,那么信号量就递减计数器并允许线程访问。

如果计数器的值为0,信号量会让线程休眠,直到计数器的值大于0。

Java的Semaphore类提供了信号量机制。


操作:

  • 初始化:传入一个参数,指有多少个信号量
  • 获取信号量:调用semaphore的acquire()方法来获取访问权限
  • 释放信号量:调用semaphore对象的release()方法释放信号量

使用信号量实现临界区来保护共享资源所要遵循的3个步骤:

  • 调用acquire()方法获得信号量。
  • 操作共享资源。
  • 调用release()方法来释放信号量。

可以一次性获取或者释放多个信号量:

acquire()、acquireUninterruptibly()、tryAcquire()以及release()方法,均有一个带有一个整型参数的版本。


Semaphore公平和非公平模式:

Semaphore类提供了重载的构造器,它允许传入第二个参数。

如果传入参数为false,则创建的Semaphore将工作在非公平模式下,这也是不传该参数时的默认模式;如果传入参数为true,则创建的Semaphore将工作在公平模式下。

倒计时机制

Java并发API提供了一个类CountDownLatch,它可以使多个线程等待直到一组操作完成。

当一个线程希望等待这些操作执行完成时,可以调用CountDownLatch对象的await(),这个方法会将调用线程休眠,直到所等待的操作全部结束。而当一个操作结束时,应该调用CountDownLatch对象的countDown()方法,该方法会将CountDownLatch对象内部的属性计数器的值减1,表示一个操作的完成。当CountDownLatch对象内部的计数器的值为0时,表示所有操作都完成了。这时,CountDownLatch对象将唤起所有因调用其await()方法而休眠的线程。


操作:

  • 初始化:一个整型参数,该参数代表了线程希望等待的操作的个数。
  • await() 方法:希望等待所有事件结束的线程需要调用该方法。
  • countDown()方法:事件在结束执行后调用。

运行模式:

在创建CountDownLatch对象时,构造器的参数将用来初始化对象内部的计数器。每次调用countDown()方法时,CountDownLatch对象的计数器的值都减1。当计数器的值为0时,CountDownLatch对象唤醒全部因调用await()方法而休眠等待的线程。


CountDownLatch一次性特性:

其实例对象的同步作用是一次性的。正如之前所示,当CountDownLatch对象内部的计数器的值为0时,后续对其的调用将不会产生任何作用。

在指定状态点同步任务

CyclicBarrier运行机制:

CyclicBarrier类的构造器需要有一个整型参数,这个参数表示在指定点进行同步的线程个数。当需要同步的线程运行到指定点时,可以调用CyclicBarrier对象的await()方法,然后等待其他线程达到指定点。这个方法使调用线程休眠,等待其他线程的到达。当最后一个需要同步的线程到达并调用CyclicBarrier对象的await()方法时,所有在此等待的线程都会被唤醒并继续执行。

CyclicBarrier类定义了内部计数器,它对需要在同步状态点进行同步的线程数进行控制。每当一个线程执行到同步状态点时,它会通过调用await()通知CyclicBarrier对象有一个线程已达到同步状态点,这时CyclicBarrier对象将内部计数器的值减1,并将调用线程休眠,直到所有线程均达到同步状态点。


构造函数传入第二个参数Runnable:

CyclicBarrier对象会在所有同步线程到达指定点后,将Runnable对象当作一个线程对象来执行。


一些其他方法:

getNumberWaiting()方法:该方法会返回因调用await()而休眠等待的线程数;

getParties()方法:该方法返回CyclicBarrier对象同步的任务数。


CyclicBarrier可以重置,这是和CountDownLatch很不同的点:

CyclicBarrier类提供的reset()方法可以实现重置。当调用这个方法后,所有因调用await()方法而休眠等待的线程,将收到BrokenBarrierException异常。


CyclicBarrie损坏状态:

当多个线程因调用await() 方法而等待时,若其中一个被中断了,则此线程会收到Interrupted- Exception异常,而其他线程将收到BrokenBarrierException异常,并且CyclicBarrier对象会进入损坏状态。

CyclicBarrier类提供了isBroken()方法。如果CyclicBarrier对象处于损坏状态,则调用该方法将返回true;否则,将返回false。

运行阶段性并发任务

Phaser类提供了在每步结束时同步线程的机制,这使得只有当所有线程都完成第一步后,才会有线程开始执行第二步操作。

CyclicBarrier也可以完成类似的功能。


Phaser对象可以动态增加和减少同步的任务数量:

在初始化Phaser对象时,必须指明参与同步的任务数量,但是它可以动态地增加或者减少同步参与者。


操作:

  • 初始化:传入一个参数,代表需要同步的线程数
  • arriveAndAwaitAdvance()方法:当一个线程调用该方法时,Phaser对象将当前阶段需要等待的线程数减1,并使调用线程休眠,直到所有线程均已完成当前阶段的任务。
  • arriveAndDeregister()方法:通知Phaser对象来减少一个参与同步的线程
  • isTerminated()方法:默认情况下(下一节将介绍非默认情况),当一个Phaser对象上只剩0个同步参与者时,Phaser对象进入一个终止状态(Termination),并且isTerminated()方法会返回true。

与其他同步工具类不同,由Phaser休眠的线程不会响应中断,也不会抛出中断异常


arrive()方法:

该方法将通知Phaser类,一个参与者已经完成当前的阶段任务,并将继续执行,不等待该阶段的其他参与者。

一个值得思考的问题:arrive之后再次arriveAndAwaitAdvance()是等待下一阶段,还是这一阶段?

答:phaser的机制很简单,每有一个线程调用了arrive,phaser需要等待的线程就减1,有个线程连续调用arrive,他甚至可以用一个线程完成一个阶段所有需要调用的线程


动态调整Phaser中参与者数量:

  • register():该方法向Phaser添加一个新的参与者,并且,默认该参与者还未完成当前阶段。
  • bulkRegister(int parties):该方法向Phaser对象增加指定数量的参与者,并且,默认这些参与者还未完成当前阶段。
  • Phaser类仅提供了一个动态减少同步参与者数量的方法:arriveAndDeregister()。

强制终止Phaser:

Phaser类提供的forceTermination()方法可改变Phaser对象的状态,不论该对象注册了多少个同步参与者,该方法都会将Phaser对象置于终止状态。

当一个Phaser对象处于终止状态时,调用awaitAdvance()和arriveAndAwait- Advance()方法将立即返回一个负数(通常状态下,返回的是正数)。

阶段性并发任务的终止状态转变控制

Phaser类提供了方法onAdvance()【protected方法】,并在阶段转变时自动调用此方法。如果该方法返回为True,则phaser进入终止状态。

onAdvance()方法的默认实现为,如果Phaser上注册的参与者数量为0,则返回true,否则为false。


需要重写该方法,实现阶段转变控制:

该方法接收刚执行完成的阶段数(从0开始计数),它还接收一个代表参与者数量的参数。

两个并发任务间的数据交换

Exchanger工具类允许在两个线程间定义一个同步点,当两个线程到达该同步点时,它们能够交换内部的数据结构,使得第一个线程的数据结构传递给第二个线程,反之亦然。

注:Exchanger只能在两个线程间同步


构造方法:

Exchanger<List<String>> exchanger=new Exchanger<>();

运行机制:

与其他同步工具类一样,第一个调用exchange()方法的线程将会休眠,直到该方法由另一个线程调用。然后同时交换数据,类似于线程间的swap操作。

★CompletableFuture机制★

CompletableFuture实现了Future接口CompletionStage接口

  • 作为Future对象,CompletableFuture会在将来返回一个结果。
  • 作为CompletionStage对象,可以在一个或多个CompletableFuture对象完成任务以后,执行额外的异步任务。

3种使用CompletableFuture类的方法:

  • 主动创建一个CompletableFuture对象,使其作为两个任务之间的同步点。一个任务创建一个值,并作为Completable对象的complete()方法的参数,该值将由CompletableFuture返回,另一个任务可以调用get()或join()方法等待该值。
  • 通过CompletableFuture类的静态方法runAsync()和supplyAsync()来执行Runnable或Supplier。这些方法将会返回一个CompletableFuture对象,并且在这些任务结束运行以后,该对象将进入完备态。在第二种情况下,Supplier的返回值即为CompletableFuture对象完备态的值。
  • 附加声明的任务将会在一个或多个CompletableFuture对象执行完毕后异步执行。该任务可以实现Runnable、Function、Consumer或BiConsumer接口。

该节的案例非常经典,写得很好,值得总结与回顾


CompletableFuture类的静态方法allOf():

该方法接收变长的CompletableFuture对象作为参数,不会等待所有传入CompletableFuture完成,才返回,不会阻塞线程


“其他说明部分”也是很经典,值得总结

原文地址:https://www.cnblogs.com/doubest/p/15488826.html