Java 同步工具类

文章目录

- CountDownLatch
- Exchanger
- CyclicBarrier
- Semaphore

写在前面

同步辅助工具类的目的是在于多线程间的协调与通信。本文参考官方文档。

CountDownLatch

允许一个或多个线程等待，直到在其它线程中执行的一组操作完成。

CountDownLatch

是用给定的

count

初始化的。由于调用了

countDown()

方法，

await

方法阻塞，直到当前计数为零之后释放所有等待线程；任何后续的

await

调用将立即返回。这是一种一次性现象——计数无法重置。如果需要重置计数的版本，可以考虑使用

CyclicBarrier

。

用例：实现多个工作线程将在同一时刻开始工作，而主线程需要等待所有工作线程完成才能继续执行。我们将使用两个

CountDownLatch

：

第一个是启动信号，它阻止任何工作线程执行，直到主线程准备好让它们开始工作；
第二个是完成信号，它使主线程等待，直到所有工作线程完成；

Worker.java：工作线程类

public class Worker implements Runnable {

    private final CountDownLatch startSignal;

    private final CountDownLatch doneSignal;

    Worker(CountDownLatch startSignal, CountDownLatch doneSignal){
        this.startSignal = startSignal;
        this.doneSignal = doneSignal;
    }

    @Override
    public void run() {
        try {
            startSignal.await();
            doWork();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            doneSignal.countDown();
        }
    }

    private void doWork(){
        System.out.println("我完成工作了！");
    }
}

Driver.java：启动类（主线程类）

public class Driver {

    public static void main(String[] args) throws InterruptedException{
        int workerNumber = 10;
        CountDownLatch startSignal = new CountDownLatch(1);
        CountDownLatch doneSignal = new CountDownLatch(workerNumber);

        for (int i = 0; i < 10 ; i++) {
            new Thread(new Worker(startSignal, doneSignal)).start();
        }

        System.out.println("工作都准备完毕！");
        System.out.println("开始执行!");
        startSignal.countDown();
        doneSignal.await();
        System.out.println("一共：" + workerNumber + "人都执行完了");
    }
}

运行将得到以下结果：

工作都准备完毕！
开始执行!
我完成工作了！
我完成工作了！
我完成工作了！
我完成工作了！
我完成工作了！
我完成工作了！
我完成工作了！
我完成工作了！
我完成工作了！
我完成工作了！
一共：10人都执行完了

await 方法在调用时，计数已经变为0，该方法将立即返回。

发现大多数博客都未对

await()

方法的

InterruptedException

异常进行解释，该异常是指当前线程的

interrupt()

方法被调用，会导致该异常的抛出；我们可以在调用

await()

方法时，捕获该异常，届时线程的中断状态将会被清除。

CountDownLatch

的

await()

可以设置时间参数，该方法拥有返回值，如果在指定的时间内得以继续执行，该方法将返回

true

, 如果是超时，该方法将放回 false，程序接着往下执行，该方法依然能够响应中断异常。

有一种大家都在遵守的模式，如果在调用 await() 超时的方法时，需要有条件的判断，那么应基于循环的条件判断，而不是 if，例如，我有着这样一个条件，当商品总数大于 10000 时，信号 CountDownLatch 将等待，我们应该这样调用：
while(total > 10000){
    signalCountDownLatch.await(1, TimeUnit.SECONDS);
}

// 不是这样，尽管我们很容易写成这样
if(total > 10000){
    signalCountDownLatch.await(1, TimeUnit.SECONDS);
}
           

Exchanger

线程间的元素交换。每个线程在进入

exchange

方法时，传递某个对象，等待与合作伙伴线程匹配，并在返回时接收其合作伙伴的对象。交换器可以看作是同步队列的双向形式。

用例：使用交换器在线程之间交换缓冲区，以便填充缓冲区的线程在需要时得到一个新清空的缓冲区，并将填充的缓冲区交给正在清空缓冲区的线程。

FillAndEmpty.java

package com.duofei.synchron;

import java.nio.ByteBuffer;
import java.util.concurrent.Exchanger;

/**
 * Exchanger 同步工具学习
 * @author duofei
 * @date 2019/11/20
 */
public class FillAndEmpty {

    Exchanger<ByteBuffer> exchanger = new Exchanger<>();

    ByteBuffer initialEmptyBuffer = ByteBuffer.allocateDirect(10);
    ByteBuffer initialFullBuffer = ByteBuffer.allocateDirect(10);

    public static void main(String[] args) {
        FillAndEmpty fillAndEmpty = new FillAndEmpty();
        new Thread(fillAndEmpty.new FillingLoop()).start();
        new Thread(fillAndEmpty.new Emptying()).start();

    }

    class FillingLoop implements Runnable{

        @Override
        public void run() {
            ByteBuffer currentBuffer = initialEmptyBuffer;
            try {
                currentBuffer.clear();
                while (currentBuffer != null){
                    if(currentBuffer.hasRemaining()){
                        currentBuffer.put(new String("he").getBytes());
                    }
                    if(!currentBuffer.hasRemaining()){
                        currentBuffer = exchanger.exchange(currentBuffer);
                        currentBuffer.clear();
                        System.out.println("重新获得空的缓冲区");
                    }
                }
            }catch (InterruptedException ex){
                ex.printStackTrace();
            }

        }
    }

    class Emptying implements Runnable{

        @Override
        public void run() {
            ByteBuffer currentBuffer = initialFullBuffer;
            try{
                currentBuffer.flip();
                byte[] bytes = new byte[10];
                while (currentBuffer != null){
                    if (currentBuffer.hasRemaining()){
                        currentBuffer.get(bytes);
                        System.out.println("缓存区内容为：" + new String(bytes));
                    }
                    if(!currentBuffer.hasRemaining()){
                        currentBuffer = exchanger.exchange(currentBuffer);
                        currentBuffer.flip();
                        System.out.println("重新获得拥有数据的缓冲区");
                    }
                }
            }catch (InterruptedException ex){
                ex.printStackTrace();
            }
        }
    }
}

控制台打印内容：

重新获得空的缓冲区
重新获得拥有数据的缓冲区
缓存区内容为：hehehehehe

上述语句在控制台会成块（上面的三条打印语句）打印，但块里的语句打印顺序不定。

CyclicBarrier

允许一组线程彼此等待到达一个共同的障碍点。

cyclicbarrier

在包含固定大小的线程的程序中非常有用，这些线程有时必须彼此等待。这个屏障被称为循环屏障，因为它可以在等待的线程被释放后重新使用。仅从这里来看，它的特性和

CountDownLatch

（计数为1的）类似，唯一的差别在于

CountDownLatch

中的计数无法被重置，而

cyclicbarrier

中的状态是可以重用的。

其实它们是不同的，或许描述为 " CountDownLatch 计数为1 的情况下能完成部分栅栏的工作" 更为合理；并且，

CyclicBarrier

在构造时支持一个可选的Runnable命令，该命令在每一个屏障点上运行一次，在最后一个线程到达之后，但是在释放任何线程之前。这个屏障动作对于在任何一方继续之前更新共享状态非常有用。

用例：使用屏障实现并行分解。每个工作线程处理矩阵的一行，然后在屏障处等待，直到处理完所有行。在处理所有行时，执行所提供的Runnable barrier操作并合并行。

Solver.java

package com.duofei.synchron;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
 * CyclicBarrier
 * @author duofei
 * @date 2019/11/20
 */
public class Solver {

    final int N;
    final float[][] data;
    final CyclicBarrier barrier;

    public static void main(String[] args) {
        Solver solver = new Solver(new float[10][2]);

        for (int i = 0; i < solver.N; i++) {
            new Thread(solver.new Worker(i)).start();
        }
    }

    class Worker implements Runnable{

        int myRow;
        Worker(int row){
            myRow = row;
        }

        @Override
        public void run() {
            System.out.println("我处理完成第" + myRow + "行");
            try {
                barrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

    public Solver(float[][] matrix){
        data = matrix;
        N = matrix.length;
        barrier = new CyclicBarrier(N, ()->{
            System.out.println("行数据合并");
        });
    }
}

输出结果：

我处理完成第0行
我处理完成第1行
我处理完成第3行
我处理完成第2行
我处理完成第4行
行数据合并
我完成了4行，我是第0个到达屏障处的
我完成了0行，我是第4个到达屏障处的
我完成了1行，我是第3个到达屏障处的
我完成了3行，我是第2个到达屏障处的
我完成了2行，我是第1个到达屏障处的

还有一点，在调用

await

方法时，还抛出

BrokenBarrierException

方法，这是什么样的情形呢？官方文档描述到：

如果一个线程由于中断,失败,或超时，过早离开障碍点；所有在障碍点等待的线程也将离开，通过捕获的 BrokenBarrierException 异常 ( 或 InterruptedException 如果他们也被打断了大约在同一时间)。

由于异常的原因离开屏障，并不会执行构造屏障时的

Runnable

，并且当一个线程在已经破坏了的屏障前等待时，会立即得到一个

BrokenBarrierException

异常。我们可以修改上述代码来验证我们的猜想：

package com.duofei.synchron;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
 * CyclicBarrier
 * @author duofei
 * @date 2019/11/20
 */
public class Solver {

    final int N;
    final float[][] data;
    final CyclicBarrier barrier;

    public static void main(String[] args) {
        Solver solver = new Solver(new float[5][2]);

        for (int i = 0; i < solver.N; i++) {
            final Thread thread = new Thread(solver.new Worker(i));
            thread.start();
            System.out.println(System.currentTimeMillis() + ": 打断" + thread.getName());
            // 测试线程打断
            thread.interrupt();
        }
    }

    class Worker implements Runnable{

        int myRow;
        Worker(int row){
            myRow = row;
        }

        @Override
        public void run() {
            System.out.println("我处理完成第" + myRow + "行");
            try {
                final int await = barrier.await();
                System.out.println("我完成了" + myRow+ "行，我是第" + await + "个到达屏障处的");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                System.out.println(System.currentTimeMillis() + "屏障破坏异常：" + Thread.currentThread().getName());
                e.printStackTrace();
            }
        }
    }

    public Solver(float[][] matrix){
        data = matrix;
        N = matrix.length;
        barrier = new CyclicBarrier(N, ()->{
            System.out.println("行数据合并");
        });
    }
}

尽管我新增了很多打印语句，但由于程序执行的非常快，这仍然很难判断，但通过基于线程的断点调试，你将会获得更清楚的结果。执行上述语句，我们将捕获到四个

BrokenBarrierException

异常，一个中断异常，构造屏障时的

Runnable

也并未执行。

Semaphore

计数信号量。从概念上讲，信号量维护一组许可证。如果需要，每个

acquire()

都会阻塞，直到获得许可证，然后获取许可证。每个

release()

添加一个许可证，潜在地释放一个阻塞的获取者。但是，没有实际使用许可证对象;信号量只是保持可用数量的一个计数，并相应地进行操作。

用例：信号量通常用于限制能够访问某些(物理或逻辑)资源的线程数量。例如，这里有一个类使用信号量来控制对池中数据的处理。

Pool.java

package com.duofei.synchron;

import java.util.concurrent.Semaphore;
import java.util.function.Consumer;

/**
 * semaphore 学习
 * @author duofei
 * @date 2019/11/20
 */
public class Pool {

    private static final int MAX_AVAILABLE = 5;
    private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);

    private Integer data = new Integer(10);

    /**
     * 处理池中数据
     */
    int handleData(Consumer<Integer> handle){
        try {
            available.acquire();
            handle.accept(data);
            return data.intValue();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return -1;
    }

    /**
     * 释放池中数据
     */
    void releaseData(){
        available.release();
    }

    public static void main(String[] args) {
        Pool pool = new Pool();
        for (int i = 0; i < 10; i++) {
            new Thread(()-> pool.handleData(System.out::println)).start();
            // 处理完数据释放
            // pool.releaseData();
        }
    }
}

在不释放数据时，我们仅仅只能处理五次数据，剩下的线程并没有处理数据的机会。

一个初始化为一个的信号量，它的使用方式是最多只有一个可用的许可证，可以用作互斥锁。这通常被称为二进制信号量，因为它只有两种状态:一个可用许可证，或者0个可用许可证。当以这种方式使用时，二进制信号量具有这样的属性(与许多锁实现不同)，即“锁”可以由所有者以外的线程释放(因为信号量没有所有权的概念)。这在某些特定上下文中很有用，比如死锁恢复。

该类的构造函数可选地接受公平性参数。当设置为
false 时，该类不能保证线程获得许可的顺序。特别是，允许倒挂，也就是说，可以一直在等待的线程之前为调用acquire()的线程分配许可证——从逻辑上讲，新线程将自己放在等待线程队列的最前面。当公平性设置为真时，信号量保证会选择调用任何 acquire 方法的线程，从而按照它们对这些方法的调用的处理顺序(先进先出)。请注意，FIFO顺序必然适用于这些方法中的特定内部执行点。因此，一个线程可以在另一个线程之前调用 acquire ，但是在另一个线程之后到达排序点，同样地，在方法返回时到达排序点。还请注意，不定时 tryAcquire 方法不尊重公平设置，但将采取任何许可是可用的。

通常，用于控制资源访问的信号量应该被初始化为公平的，以确保没有线程因为访问资源而饿死。当将信号量用于其他类型的同步控制时，非公平排序的吞吐量优势常常超过了公平性考虑。

这个类还提供了一次获取和释放多个许可的便利方法。当使用这些方法而不将公平性设置为真时，要注意无限期延迟的风险。

官方文档的解释非常棒，如果你认真品的话。

Java 同步工具类

文章目录

CountDownLatch

Exchanger

CyclicBarrier

Semaphore

继续阅读

关于Gradle配置的小结

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method