zoukankan html css js c++ java

并发工具类使用详解及区别（CountDownLatch、CyclicBarrier、Semaphore、Exchanger）

本文转载自：码农历险记 CountDownLatch

CountDownLatch介绍

CountDownLatch是一个同步工具类，它允许一个或多个线程一直等待，直到其他线程执行完后再执行。例如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有框架服务之后执行。

CountDownLatch 是通过一个计数器来实现的，计数器的初始化值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就相应得减1。当计数器到达0时，表示所有的线程都已完成任务，然后在闭锁上等待的线程就可以恢复执行任务。

CountDownLatch原理示意图

CountDownLatch的伪代码：

Main thread start
Create CountDownLatch for N threads
Create and start N threads
Main thead wait on latch
N threads completes there tasks are returns
Main thread resume execution

CountDownLatch.java中定义的构造函数:

//用等待的线程数量来进行初始化
public void CountDownLatch(int count){...}

计数器count是闭锁需要等待的线程数量，只能被设置一次，且CountDownLatch没有提供任何机制去重新设置计数器count。

与CountDownLatch的第一次交互是主线程等待其他线程。主线程必须在启动其他线程后立即调用CountDownLatch.await()方法。这样主线程的操作就会在这个方法上阻塞，直到其他线程完成各自的任务。

其他N个线程必须引用CountDownLatch闭锁对象，因为它们需要通知CountDownLatch对象，它们各自完成了任务；这种通知机制是通过CountDownLatch.countDown()方法来完成的；每调用一次，count的值就减1，因此当N个线程都调用这个方法，count的值就等于0，然后主线程就可以通过await()方法，恢复执行自己的任务。

在实时系统中的使用场景

实现最大的并行性：有时我们想同时启动多个线程，实现最大程度的并行性。例如，我们想测试一个单例类。如果我们创建一个初始计数器为1的CountDownLatch，并让其他所有线程都在这个锁上等待，只需要调用一次countDown()方法就可以让其他所有等待的线程同时恢复执行。
开始执行前等待N个线程完成各自任务：例如应用程序启动类要确保在处理用户请求前，所有N个外部系统都已经启动和运行了。
死锁检测：一个非常方便的使用场景是你用N个线程去访问共享资源，在每个测试阶段线程数量不同，并尝试产生死锁。

CountDownLatch使用例子

模拟一个应用程序启动类，开始就启动N个线程，去检查N个外部服务是否正常并通知闭锁；启动类一直在闭锁上等待，一旦验证和检查了所有外部服务，就恢复启动类执行。

BaseHealthChecker.java :这个类是实现了Runnable接口，负责所有特定的外部服务健康检查的基类。

import java.util.concurrent.CountDownLatch;

public abstract class BaseHealthChecker implements Runnable {

    private CountDownLatch _latch;
    private String _serviceName;
    private boolean _serviceUp;

    public BaseHealthChecker(String serviceName, CountDownLatch latch)
    {
        super();
        this._latch = latch;
        this._serviceName = serviceName;
        this._serviceUp = false;
    }

    @Override
    public void run() {
        try {
            verifyService();
            _serviceUp = true;
        } catch (Throwable t) {
            t.printStackTrace(System.err);
            _serviceUp = false;
        } finally {
            if(_latch != null) {
                _latch.countDown();
            }
        }
    }

    public String getServiceName() {
        return _serviceName;
    }

    public boolean isServiceUp() {
        return _serviceUp;
    }

    public abstract void verifyService();
}

NetworkHealthChecker.java,DatabaseHealthChecker.java和CacheHealthChecker.java都继承自BaseHealthChecker，引用CountDownLatch实例，除了服务名和休眠时间不同外，都实现各自的verifyService方法。

NetworkHealthChecker.java类

import java.util.concurrent.CountDownLatch;

public class NetworkHealthChecker extends BaseHealthChecker
{
    public NetworkHealthChecker (CountDownLatch latch)
    {
        super("Network Service", latch);
    }

    @Override
    public void verifyService() 
    {
        System.out.println("Checking " + this.getServiceName());
        try 
        {
            Thread.sleep(7000);
        } 
        catch (InterruptedException e)
        {
            e.printStackTrace();
        }
        System.out.println(this.getServiceName() + " is UP");
    }
}

DatabaseHealthChecker.java类

import java.util.concurrent.CountDownLatch;

public class DatabaseHealthChecker extends BaseHealthChecker
{
    public DatabaseHealthChecker (CountDownLatch latch)
    {
        super("Database Service", latch);
    }

    @Override
    public void verifyService() 
    {
        System.out.println("Checking " + this.getServiceName());
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(this.getServiceName() + " is UP");
    }
}

CacheHealthChecker.java类

import java.util.concurrent.CountDownLatch;

public class CacheHealthChecker extends BaseHealthChecker
{
    public CacheHealthChecker (CountDownLatch latch)
    {
        super("Cache Service", latch);
    }

    @Override
    public void verifyService() 
    {
        System.out.println("Checking " + this.getServiceName());
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(this.getServiceName() + " is UP");
    }
}

ApplicationStartupUtil.java:是一个主启动类，它负责初始化闭锁，然后等待所有服务都被检查完成，再恢复执行。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class ApplicationStartupUtil 
{
    private static List<BaseHealthChecker> _services;
    private static CountDownLatch _latch;

    private ApplicationStartupUtil()
    {
    }

    private final static ApplicationStartupUtil INSTANCE = new ApplicationStartupUtil();

    public static ApplicationStartupUtil getInstance()
    {
        return INSTANCE;
    }

    public static boolean checkExternalServices() throws Exception
    {
        _latch = new CountDownLatch(3);
        _services = new ArrayList<BaseHealthChecker>();
        _services.add(new NetworkHealthChecker(_latch));
        _services.add(new CacheHealthChecker(_latch));
        _services.add(new DatabaseHealthChecker(_latch));

        Executor executor = Executors.newFixedThreadPool(_services.size());

        for(final BaseHealthChecker v : _services) 
        {
            executor.execute(v);
        }

        _latch.await();

        for(final BaseHealthChecker v : _services) 
        {
            if( ! v.isServiceUp())
            {
                return false;
            }
        }
        return true;
    }
}

测试代码检测闭锁功能：

public class Main {
    public static void main(String[] args) 
    {
        boolean result = false;
        try {
            result = ApplicationStartupUtil.checkExternalServices();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("External services validation completed !! Result was :: "+ result);
    }
}

执行结果：

Checking Network Service
Checking Cache Service
Checking Database Service
Database Service is UP
Cache Service is UP
Network Service is UP
External services validation completed !! Result was :: true

工作中的使用（优雅地完成初始化）

在移动应用开发中（以Android为例），随着功能的增多，应用初始化工作开始增多，网络，账号，推送服务，预加载数据等依次登场，开发人员都会临时在Application中找到现有初始化逻辑，将自己的代码插在其中。随着版本的迭代，新老员工的交替，几乎没人能对应用的初始化过程完全了解，删除一行初始化代码甚至移动位置都可能造成严重的后果。

应用初始化过程极其重要，它是应用后续平稳运行的前提和保证。开发初始化配置模块（公司内部开源不宜公开），更好地管理初始化逻辑，对初始化地工作进行分层，分优先级，多线程地规划，进而在大幅提升初始化效率，同时还有完整地日志监控体系功能。有了它，规划整个初始化工作将简单而优雅.

同步工具类可以是任何一个对象，只要它根据其自身的状态来协调线程的控制流。阻塞队列可以作为同步工具类，其他类型的同步工具类还包括信号量（Semaphore）、栅栏（Barrier）以及闭锁（Latch）。本文就目前常用的3种同步工具类进行简单介绍。

闭锁

闭锁是一种同步工具类，可以延迟线程的进度直到其到达终止状态。闭锁的作用相当于一扇门：在闭锁到达结束状态之前，这扇门一直是关闭的，并且没有任何线程能够通过，当到达结束状态时，这扇门会打来并允许所有的线程通过。当闭锁到达结束状态后，将不会再改变状态，因此这扇门将永远保持打开状态。闭锁可以用来确保某些活动直到其他活动都完成后才继续执行，例如：

确保某个计算在其需要的所有资源都被初始化之后才继续执行。
确保某个服务在其依赖的所有其他服务都已经启动之后才启动。
等到直到直到某个操作的所有参与者（例如，在多玩家游戏中的所有玩家）都就绪再继续执行。

CountDownLatch是一种灵活的闭锁实现。闭锁状态包括一个计数器，该计数器被初始化为一个正数，表示需要等待的事件数量。countDown方法递减计数器，表示有一个事件已经发生了，而await方法等待计数器达到零，这表示所有需要等待的事件都已发生。如果计数器的值非零，那么await方法会一直阻塞直到计算器为零，或者等待中的线程中断，或者超时。

一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。

用给定的计数初始化 CountDownLatch。由于调用了 countDown() 方法，所以在当前计数到达零之前，await 方法会一直受阻塞。之后，会释放所有等待的线程，await 的所有后续调用都将立即返回。这种现象只出现一次计数无法被重置。如果需要重置计数，请考虑使用 CyclicBarrier。

CountDownLatch 是一个通用同步工具，它有很多用途。将计数 1 初始化的 CountDownLatch 用作一个简单的开/关锁存器，或入口：在通过调用 countDown() 的线程打开入口前，所有调用 await 的线程都一直在入口处等待。用 N 初始化的 CountDownLatch 可以使一个线程在 N 个线程完成某项操作之前一直等待，或者使其在某项操作完成 N 次之前一直等待。

CountDownLatch 的一个有用特性是，它不要求调用 countDown 方法的线程等到计数到达零时才继续，而在所有线程都能通过之前，它只是阻止任何线程继续通过一个 await。

下面是CountDownLatch的一个简单例子：运行代码看效果更容易理解


package com.joonwhee.imp;


import java.util.concurrent.CountDownLatch;

import java.util.concurrent.TimeUnit;


/**

* CountDownLatch的简单例子

* @author 

* @Date 2018年1月27日

*/

public class CountDownLatchTest {


static CountDownLatch timeOutCountDownLatch = new CountDownLatch(1);


public static void main(String args[]) {

try {

new Driver(10);

testAwaitTimtOut();

} catch (InterruptedException e) {

e.printStackTrace();

}

}


// 测试带超时的await方法

public static void testAwaitTimtOut() throws InterruptedException {

System.out.println("before await(long timeout, TimeUnit unit)");

timeOutCountDownLatch.await(3, TimeUnit.SECONDS); //等待超时时间为3秒

System.out.println("after await(long timeout, TimeUnit unit)");

}


}


class Driver {

public Driver(int N) throws InterruptedException {

CountDownLatch startSignal = new CountDownLatch(1); // 定义一个CountDownLatch, 计数器值为1, 也就是每次await(), 需要执行1次countDown(), 才能继续执行await()外面的代码

CountDownLatch doneSignal = new CountDownLatch(N); // 定义一个CountDownLatch, 计数器值为N, 也就是每次await(), 需要执行N次countDown(), 才能继续执行await()外面的代码


for (int i = 0; i < N; ++i) {

// 创建并启动线程

new Thread(new Worker(startSignal, doneSignal)).start();

}

Thread.sleep(2000); // 睡眠2秒, 可以看到10个线程都在等待startSignal.countDown()执行

System.out.println();

startSignal.countDown(); // 解除所有线程的阻塞

doneSignal.await(); // 等待所有线程执行doneSignal.countDown(), 才通过

System.out.println();

System.out.println("Main thread-after:doneSignal.await()------");

}

}


class Worker implements Runnable {

private final CountDownLatch startSignal;

private final CountDownLatch doneSignal;


Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {

this.startSignal = startSignal;

this.doneSignal = doneSignal;

}


public void run() {

try {

System.out.println(Thread.currentThread().getName() + "-before:startSignal.await()");

startSignal.await(); // 线程会在此处等待, 直到startSignal.countDown()执行

System.out.println(Thread.currentThread().getName() + "-after:startSignal.await()");

doneSignal.countDown();

} catch (InterruptedException ex) {

} // return;

}

}

用CountDownLatch


import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class Demo2 {

    private int[] nums;

    public Demo2(int line) {
        nums = new int[line];
    }

    public void calc(String line, int index, CountDownLatch latch) {
        String[] nus = line.split(","); // 切分出每个值
        int total = 0;
        for (String num : nus) {
            total += Integer.parseInt(num);
        }
        nums[index] = total; // 把计算的结果放到数组中指定的位置
        System.out.println(Thread.currentThread().getName() + " 执行计算任务... " + line + " 结果为：" + total);
        latch.countDown();
    }

    public void sum() {
        System.out.println("汇总线程开始执行... ");
        int total = 0;
        for (int i = 0; i < nums.length; i++) {
            total += nums[i];
        }
        System.out.println("最终的结果为：" + total);
    }

    public static void main(String[] args) {

        List<String> contents = readFile();
        int lineCount = contents.size();

        CountDownLatch latch = new CountDownLatch(lineCount);

        Demo2 d = new Demo2(lineCount);
        for (int i = 0; i < lineCount; i++) {
            final int j = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    d.calc(contents.get(j), j, latch);
                }
            }).start();
        }

        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        d.sum();
    }

    private static List<String> readFile() {
        List<String> contents = new ArrayList<>();
        String line = null;
        BufferedReader br = null;
        try {
            br = new BufferedReader(new FileReader("d:\nums.txt"));
            while ((line = br.readLine()) != null) {
                contents.add(line);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (br != null) {
                try {
                    br.close();
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }
        return contents;
    }

}

这里写图片描述

这里写图片描述
这里其实就是把传统的做法变化了一下、CountDownLatch利用一个计数器来完成，这里面最主要的方法就是上面图片中所标识的2个方法、在上述实例中先完成行数的统计传入CountDownLatch，在线程每完成一个就调用CountDownLatch的countDown方法来将数量减一、达到零时，则释放所有等待线程。

CountDownLatch实现原理

CountDownLatch是继承了AbstractQueuedSynchronizer类，一个共享锁的实现
这里写图片描述

上图的await方法就是判断getState的值是否等于0等于0执行不等于0返回-1所以不执行

栅栏

栅栏（Bariier）类似于闭锁，它能阻塞一组线程知道某个事件发生。栅栏与闭锁的关键区别在于，所有的线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待等待时间，而栅栏用于等待线程。栅栏用于实现一些协议，例如几个家庭决定在某个地方集合：“所有人6：00在麦当劳碰头，到了以后要等其他人，之后再讨论下一步要做的事情”。

CyclicBarrier 可以使一定数量的参与方反复的在栅栏位置汇聚，它在并行迭代算法中非常有用：将一个问题拆成一系列相互独立的子问题。当线程到达栅栏位置时，调用await() 方法，这个方法是阻塞方法，直到所有线程到达了栅栏位置，那么栅栏被打开，此时所有线程被释放，而栅栏将被重置以便下次使用。如果对await的调用超时，或者await阻塞的线程被中断，那么栅栏就认为是打破了，所有阻塞的await调用都将终止并抛出BrokenBarrierException。CycleBarrier还可以使你将一个栅栏操作传递给构造函数，这是一个Runnable，当成功通过栅栏会（在一个子任务线程中）执行它，但在阻塞线程被释放之前是不能执行的。

下面是CycleBarrier的一个简单例子：运行代码看效果更容易理解


package com.joonwhee.imp;


import java.util.concurrent.BrokenBarrierException;

import java.util.concurrent.CyclicBarrier;


/**

* CyclicBarrier的简单例子

* @author 

* @Date 2018年1月27日

*/

class Solver {

final CyclicBarrier barrier;


class Worker implements Runnable {

public void run() {

try {

System.out.println(Thread.currentThread().getName() + ":before barrier.await()");

Thread.sleep(1000); // 睡眠1秒, 便于更好的观察

barrier.await(); // 当所有线程到达此处时, 栅栏打开, 先执行定义栅栏时自带的Runnable方法, 所有线程得以往下执行

System.out.println(Thread.currentThread().getName() + ":after barrier.await()");

} catch (InterruptedException ex) {

return;

} catch (BrokenBarrierException ex) {

return;

}

}

}


public Solver(int key) throws InterruptedException {

// 定义一个CyclicBarrier, 等待的线程数为key个, 并且自带一个Runnable方法

barrier = new CyclicBarrier(key, new Runnable() {

public void run() {

try {

System.out.println("所有线程执行await()后, 执行本run方法, 也就是栅栏打开后会先执行本方法");

Thread.sleep(1000); // 睡眠1秒, 更好的观察此方法的执行顺序

} catch (InterruptedException e) {

e.printStackTrace();

}

}

});

// 第一次: 创建key个线程执行, 与上文定义CyclicBarrier的数量相同（都为key）

for (int i = 0; i < key ; ++i){

new Thread(new Worker()).start();

}

Thread.sleep(3000); // 睡眠3秒, 等待上面的所有线程执行完毕

System.out.println();

// 第二次: 创建key个线程执行, 测试栅栏是可以反复使用的

for (int i = 0; i < key ; ++i){ //

new Thread(new Worker()).start();

}

}


}

public class CyclicBarrierTest {

public static void main(String args[]) throws InterruptedException{

new Solver(10);

}

}

一个同步辅助类，它允许一组线程互相等待，直到到达某个公共屏障点 (common barrier point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 barrier 在释放等待线程后可以重用，所以称它为循环的 barrier。

CyclicBarrier 支持一个可选的 Runnable 命令，在一组线程中的最后一个线程到达之后（但在释放所有线程之前），该命令只在每个屏障点运行一次。若在继续所有参与线程之前更新共享状态，此屏障操作很有用。
这里写图片描述
如：一个开会的例子、不管你什么时候到的，一定要等到规定人数到了之后才进行、否则就一直等待，直到人数达到规定数量才继续向下执行


import java.util.Random;
import java.util.concurrent.CyclicBarrier;

public class Demo {

    Random random = new Random();

    public void meeting(CyclicBarrier barrier) {
        try {
            Thread.sleep(random.nextInt(4000));
        } catch (InterruptedException e1) {
            e1.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + " 到达会议室，等待开会..");


        try {
            barrier.await();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) {
        Demo demo = new Demo();

        CyclicBarrier barrier = new CyclicBarrier(10, new Runnable() {
            @Override
            public void run() {
                System.out.println("好！我们开始开会...");
            }
        });

        for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    demo.meeting(barrier);
                }
            }).start();
        }

    }

}

CyclicBarrier实现原理

这里写图片描述
这里的CyclicBarrier其实和CountDownLatch差不多、这里还传入了一个事件，就是当条件达到的时候需要做什么事

这个的话其实我们就是调用await方法、而await方法就调用了dowait方法，这个dowait方法就是整个CyclicBarrier类的核心

private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Generation g = generation;
            //通过Generation的broken方法了判断是否中断，如果中断了，抛出一个中断的异常
            if (g.broken)
                throw new BrokenBarrierException();
            //判断这个线程有没有被中断过
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }
            //进来一个线程，count就递减，说明等待的线程就少一个
            int index = --count;
            if (index == 0) {  // tripped
            //当index等待线程为0则叫醒所有的等待线程
                boolean ranAction = false;
                try {
                    //在判断barrierAction是否有事件传入，有就执行
                    final Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    nextGeneration();
                    return 0;
                } finally {
                    //如果执行事件报错则打破屏障叫醒所有线程
                    if (!ranAction)
                        breakBarrier();
                }
            }
            //后面就是不为0的处理判断是否重置超时什么的
            // loop until tripped, broken, interrupted, or timed out
            for (;;) {
                try {
                    if (!timed)
                        trip.await();
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                    if (g == generation && ! g.broken) {
                        breakBarrier();
                        throw ie;
                    } else {
                        // We're about to finish waiting even if we had not
                        // been interrupted, so this interrupt is deemed to
                        // "belong" to subsequent execution.
                        Thread.currentThread().interrupt();
                    }
                }

                if (g.broken)
                    throw new BrokenBarrierException();

                if (g != generation)
                    return index;

                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

这里的话引用了ReentrantLock进行加锁，来保证线程是安全的，而Generation就是为了实现reset重置功能的
这里写图片描述

信号量

计数信号量（counting semaphore）用来控制同时访问某个特定资源的操作数量，或者同时执行某个制定操作的数量。计数信号量还可以实现某种资源池，或者对容器施加边界。

Semaphore中管理着一组虚拟的许可（permit），许可的初始数量可通过构造函数来指定。在执行操作时可以首先获得许可（只要还有剩余的许可），并在使用以后释放许可。如果没有许可，那么acquire将阻塞直到有许可（或者直到被中断或者操作超时）。release方法将返回一个许可给信号量。计算信号量的一种简化形式是二值信号量，即初始值为1的Semaphore。二值信号量可以用作互斥体（mutex），并具备不可重入的加锁语义：谁拥有这个唯一的许可，谁就拥有可互斥锁。

Semaphore可以用于实现资源池，例如数据库连接池。我们可以构造一个固定长度的资源池，当池为空时，请求资源将会失败，但你真正希望看到的行为是阻塞而不是失败，并且当池非空时解除阻塞。如果将Semaphore的计数值初始化为池的大小，并在从池中获取一个资源之前首先调用acquire方法获取一个许可，在将资源返回给池之后调用release释放许可，那么acquire将一直阻塞直到资源池不为空。

下面是Semaphore的一个简单例子：运行代码看效果更容易理解

package com.joonwhee.imp;


import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Semaphore;


/**

* Semaphore的简单例子

* @author 

* @Date 2018年1月27日

*/

public class SemaphoreTest {


public static final ExecutorService THREAD_POOL = Executors.newFixedThreadPool(5);

// 定义一个信号量, 许可数量只有1, 也就是同时最多只能有1个线程访问, 可以通过修改许可数量来观察输出有什么不同

private static final Semaphore available = new Semaphore(1);


public static void doSomeThing() throws InterruptedException {

available.acquire(); // 尝试获取一个许可, 阻塞直到有一个可用, 或者被打断

// 获取许可用, 进行自己的逻辑处理, 此处只输出一句话

System.out.println(Thread.currentThread().getName() + "-doSomeThing");

Thread.sleep(1000); // 为了看的更清楚, 睡眠1秒

available.release(); // 释放许可

}


public static void main(String args[]) throws InterruptedException {

// 开启5个线程执行doSomeThing()

for (int i = 0; i < 5; i++) {

THREAD_POOL.submit(new Runnable() {

@Override

public void run() {

try {

doSomeThing();

} catch (InterruptedException e) {

e.printStackTrace();

}

}

});

}


}

}

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

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


import java.util.concurrent.Semaphore;

public class Demo {

    public void method (Semaphore semaphore) {

        try {
            semaphore.acquire();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(Thread.currentThread().getName() + " is run ...");

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        semaphore.release();
    }


    public static void main(String[] args) {

        Demo d = new Demo();
        Semaphore semaphore = new Semaphore(10);

        while(true) {

            new Thread(new Runnable() {
                @Override
                public void run() {
                    d.method(semaphore);
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }).start();
        }


    }

}

每次只也许n个线程执行、其余的线程将被等待在那不释放也不有些、直到前面n个线程有一个运行完毕后释放资源后、其余线程去竞争运行。与线程池的差不多，只不过Semaphore会创建线程等待在那、而线程池不会创建线程、都是规定只能有n个线程执行。

Exchanger

可以在对中对元素进行配对和交换的线程的同步点。每个线程将条目上的某个方法呈现给 exchange 方法，与伙伴线程进行匹配，并且在返回时接收其伙伴的对象。Exchanger 可能被视为 SynchronousQueue 的双向形式。Exchanger 可能在应用程序（比如遗传算法和管道设计）中很有用。


import java.util.concurrent.Exchanger;

public class Demo {

    public void a (Exchanger<String> exch) {

        System.out.println("a 方法执行...");

        try {
            System.out.println("a 线程正在抓取数据...");
            Thread.sleep(2000);
            System.out.println("a 线程抓取到数据...");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        String res = "12345";

        try {
            System.out.println("a 等待对比结果...  抓取数据为：" + res);
            String value = exch.exchange(res);
            System.out.println("a获取b交换过来的数据:" + value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    public void b (Exchanger<String> exch) {
        System.out.println("b 方法开始执行...");
        try {
            System.out.println("b 方法开始抓取数据...");
            Thread.sleep(4000);
            System.out.println("b 方法抓取数据结束...");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String res = "1234567";

        try {
            String value = exch.exchange(res);
            System.out.println("开始进行比对...");
            System.out.println("b获取a交换过来的数据:" + value);
            System.out.println("比对结果为：" + value.equals(res));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {

        Demo d = new Demo();
        Exchanger<String> exch = new Exchanger<>();
        new Thread(new Runnable() {
            @Override
            public void run() {
                d.a(exch);
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                d.b(exch);
            }
        }).start();

    }

}

这里写图片描述
大家可以发现，其实Exchanger作用就是线程与线程之间的数据交换

参考:

https://blog.csdn.net/qq1137623160/article/details/79768936?utm_source=copy

什么时候使用CountDownLatch
Java concurrency – CountDownLatch Example

正因为当初对未来做了太多的憧憬，所以对现在的自己尤其失望。生命中曾经有过的所有灿烂，终究都需要用寂寞来偿还。

查看全文

相关阅读:
数据库查询服务框架
 postgresql清理工具
 postgre索引
 SAP模块一句话入门(专业术语的理解)
SAP订单结算详解
 SAP Datasheet
ASP.NET MVC5 网站开发实践
 VMware S/4 HANA OP 1511虚拟机下载，64G内存限制解决方案
 SAP标准教材列表
 SAP文章链接

原文地址：https://www.cnblogs.com/candlia/p/11920054.html