7.并发工具类

作者: torres1 | 来源:发表于2020-01-09 23:23 被阅读0次

1.概述

同步工具类可以是任何一个类，只要它根据其自身的状态来协调线程的控制流。阻塞队列可以作为同步工具类，其他类型的同步工具还包括信号量（Semaphore），栅栏（Latch），闭锁(Latch).

2.闭锁

闭锁可以延迟线程的进度直到到达终止状态，闭锁的作用相当于一扇门：在闭锁到达结束状态之前，这扇门一直是关闭的，并且没有任何线程能通过，当到达结束状态时，这扇门会打开并允许所有的线程通过。
CountDownLatch
构造器：
CountDownLatch的构造函数接受一个int类型的参数作为计数器，如果你想要等待N个点完成，这里就传入N。

    public CountDownLatch(int var1) {
        if (var1 < 0) {
            throw new IllegalArgumentException("count < 0");
        } else {
            this.sync = new CountDownLatch.Sync(var1);
        }
    }

方法	说明
countDown()	每次调用该方法N就会减一
await()	在计数到达零之前，该方法会阻塞当前线程。之后会释放所有等待的线程，所有的后续调用都将立即返回。
await(long time,TimeUnit unit)	这个方法等待特定时间后，就不会阻塞当前线程。

CountDownLatch计数无法被重置。如果需要重置计数，请考虑使用CyclicBarria。由于countDown()方法可以用在任何地方，所以这里说的N个点可以是N个线程，也可以是１个线程里的N个执行步骤。

简单入门：

public class CountDownLatchTest {
    private static CountDownLatch c = new CountDownLatch(2);

    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            System.out.println(1);
            c.countDown();
            System.out.println(2);
            c.countDown();
        }).start();
        c.await();
        System.out.println(3);
    }
}

输出

1
2
3

进阶用法：
下面给出了两个类，其中一组 worker 线程使用了两个倒计数锁存器：

第一个类是一个启动信号，在 driver 为继续执行 worker 做好准备之前，它会阻止所有的 worker 继续执行。
第二个类是一个完成信号，它允许 driver 在完成所有 worker 之前一直等待。

public class Worker implements Runnable {
    private CountDownLatch startSignal;
    private CountDownLatch doneSignal;

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

    public void run() {
        try {
            startSignal.await();
            work();
            doneSignal.countDown();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void work() {
    }
}

public class Driver {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch startSignal = new CountDownLatch(1);//开始信号
        CountDownLatch doneSignal = new CountDownLatch(5);//结束信号
        for (int i = 0; i < 5; i++) {
            new Thread(
                    new Worker(startSignal, doneSignal)
            ).start();
        }
        doSomethingElse("before startSignal");
        startSignal.countDown();//开始所有工作
        doSomethingElse("before doneSignal");
        doneSignal.await();//等待所有的工作结束
        doSomethingElse("done!");
    }

    public static void doSomethingElse(String str) {
        System.out.println(str);
    }
}

输出：
before startSignal
before doneSignal
done!

3.信号量

Semaphore

一个计数信号量。从概念上讲，信号量维护了一个许可集。如有必要，在许可可用前会阻塞每一个 acquire()，然后再获取该许可。每个 release() 添加一个许可，从而可能释放一个正在阻塞的获取者。但是，不使用实际的许可对象，Semaphore 只对可用许可的号码进行计数，并采取相应的行动。

计算信号量的一种简化形式是二值信号量，即初始值为１的Semaphore，二值信号量可以用做互斥体，并具备不可重入的加锁语义。

Semaphore 通常用于限制可以访问某些资源（物理或逻辑的）的线程数目。例如:

public class SemaphoreTest {
    private static final int THREAD_COUNT = 30;
    private static ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
    private static Semaphore s = new Semaphore(10);

    public static void main(String[] args) {
        for (int i = 0; i < THREAD_COUNT; i++) {
            executorService.execute(() -> {
                try {
                    s.acquire();
                    System.out.println("save data thread=" + Thread.currentThread().getName());
                    s.release();
                } catch (InterruptedException e) {
                }
            });
        }
    }
}

代码中，虽然有30个线程在执行，但是只允许10个并发执行。Semaphore的构造方法 Semaphore(int permits)接受一个整型的数字，表示可用的许可证数量。Semaphore(10)表示允许10个线程获取许可证，也就是最大并发数是10。Semaphore的用法也很简单，首先线程使用 Semaphore的acquire()方法获取一个许可证，使用完之后调用release()方法归还许可证。还可以用tryAcquire()方法尝试获取许可证。

4.栅栏（屏障）

CyclicBarrier
字面意思是可循环使用（Cyclic）的屏障(Barrier)。它要做的事情是，让一组线程到达一个屏障（也可以叫做同步点）时阻塞，直到最后一个线程到达屏障时，屏障才会打开，所有被屏障拦截的线程才会继续执行。同时，Barrier在释放等待线程之后可以重用。

CyclicBarrier支持一个可选的Runnable命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作很有用。

栅栏与闭锁的关键区别：

闭锁是一次性对象，栅栏可以重用，也可以reset()重置。
所有线程必须都到达栅栏位置，才能继续执行。闭锁用于等待事件，而栅栏用于等待其他线程。

示例用法：
用户多线程计算数据,最后合并计算结果的场景。

一个Excel保存了用户所有的银行流水，每个Sheet保存了一个账户近一年的每笔银行流水，现在需要统计用户的日均银行流水，先用多线程处理每个sheet里的银行流水，都执行完了之后,得到每个sheet的日均银行流水,最后，再用barrierAction用这些线程的计算结果，计算整个Excel的日均银行流水。

import java.util.Map;
import java.util.concurrent.*;

/**
 * Created by dachao on 16-7-28.
 */
public class BankWaterService implements Runnable {
    /*
    创建４个屏障,处理完之后执行当前类的run方法
     */
    private CyclicBarrier c = new CyclicBarrier(4, this);
    /*
    假设只有4个sheet，所以只启动４个线程
     */
    private Executor executor = Executors.newFixedThreadPool(4);
    /*
    保存每个sheet计算出的银流结果
     */
    private ConcurrentHashMap<String, Integer> count = new ConcurrentHashMap<>();

    private void count() {
        for (int i = 0; i < 4; i++) {
            executor.execute(() -> {
                //计算当前sheet的银行数据，代码省略
                count.put(Thread.currentThread().getName(), 1);
                try {
                    c.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });

        }
    }

    @Override
    public void run() {
        int result = 0;
        //汇总每个sheet计算出的结果
        for (Map.Entry<String, Integer> sheet : count.entrySet()) {
            result += sheet.getValue();
        }
        //将结果输出
        count.put("result", result);
        System.out.println(result);
    }

    public static void main(String[] args) {
        BankWaterService service = new BankWaterService();
        service.count();
    }
}

输出结果：
4

5.线程间交换数据的Exchanger

Exchanger(交换者)
是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。这两个线程通过 exchange方法交换数据，如果第一个线程先执行exchange()方法，它会一直等待第二个线程也执行exchange方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。
Exchanger可以用于遗传算法，遗传算法里需要选出两个人作为交配对象，这时候会交换两人的数据，并使用交叉规则得出2个交配结果。Exchanger也可以用于校对工作，比如我们需要将纸制银行流水通过人工的方式录入成电子银行流水，为了避免错误，采用AB岗两人进行录入，录入到Excel之后，系统需要加载这两个Excel，并对两个Excel数据进行校对，看看是否录入一致。

import java.util.Map;
import java.util.concurrent.*;

/**
 * Created by dachao on 16-7-28.
 */
public class BankWaterService implements Runnable {
    /*
    创建４个屏障,处理完之后执行当前类的run方法
     */
    private CyclicBarrier c = new CyclicBarrier(4, this);
    /*
    假设只有4个sheet，所以只启动４个线程
     */
    private ExecutorService executor = Executors.newFixedThreadPool(4);
    /*
    保存每个sheet计算出的银流结果
     */
    private ConcurrentHashMap<String, Integer> count = new ConcurrentHashMap<>();

    private void count() {
        for (int i = 0; i < 4; i++) {
            executor.execute(() -> {
                //计算当前sheet的银行数据，代码省略
                count.put(Thread.currentThread().getName(), 1);
                try {
                    c.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });

        }
        executor.shutdown();
    }

    @Override
    public void run() {
        int result = 0;
        //汇总每个sheet计算出的结果
        for (Map.Entry<String, Integer> sheet : count.entrySet()) {
            result += sheet.getValue();
        }
        //将结果输出
        count.put("result", result);
        System.out.println(result);
    }

    public static void main(String[] args) {
        BankWaterService service = new BankWaterService();
        service.count();

    }
}

输出：
4

7.并发工具类

1.概述

2.闭锁

3.信号量

4.栅栏（屏障）

5.线程间交换数据的Exchanger

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读