Java中的并发工具类_java 并发工具类

Java中的并发工具类

1. CountDownLatch

join用于让当前执行线程等待join线程执行结束。实现原理是不停检查join线程是否存活，如果存活则让当前线程永远等待；线程中止后，线程notifyAll()会被调用(JVM)中实现；

 public static void main(String[] args) throws InterruptedException {
        final int n = 6;
        //计数为6
        CountDownLatch count = new CountDownLatch(n);
        for (int i = 0; i < n; i++) {
            new Thread(()->{
                System.out.println("第"+Thread.currentThread().getName()
                        +"位");
                count.countDown();//-1操作
            },i+"").start();
           // count.countDown();
        }

        count.await();//等到为0的时候才继续执行下去
        System.out.println("OK,操作完毕！");
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

CountDownLatch类似倒数计数器，使用countDown()方法进行减法操作，await()会阻塞当前线程，直到变成0才解除阻塞；或则设置为await(long time,TimeUnit unit),当超过time时间后不归零也会解除阻塞；

注意点：该类不可以重新初始化或修改计数器的值；

2.同步屏障CyclicBarrier

和上面那个倒数计数器相反，每遇到一个await()阻塞,屏障值+1，等加到指定的屏障值，所有的因为await()阻塞的线程阻塞都会解除；如果屏障值达不到就持续阻塞；

该构造方法有一个是CyclicBarrier(int parties,Runnable barrierAtion)，意思是达到屏障值parties后，先优先执行barrierAtion；

/**
 * 加法计数器
 * 达到一定数量的await之后，阻塞解除
 */
public class CyclicBarrierDemo {
    public static void main(String[] args) {
        final int num = 7;
        CyclicBarrier cyclicBarrier = new CyclicBarrier(num, ()->System.out.println("我优先完成！"));
        for (int i = 0; i < num; i++) {
            new Thread(()->{
                try {
                    cyclicBarrier.await();
                    System.out.println(Thread.currentThread().getName());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            },i+"").start();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

结果：

3.CountDownLatch与CyclicBarrier的区别

区别点：

(1)CountDownLatch的计数器只能使用一次，而CyclicBarrier可以使用reset()方法重置。后者适用于更为复杂的业务场景；

(2)CyclicBarrier可以使用getNumberWaiting()方法获取该类阻塞的线程数量。isBroken()方法用来了解阻塞的线程是否被中断；

4.Semaphore

信号量，用来控制同时访问特定资源的线程数量；

主要用于做流量控制，比如数据库连接；

用acquire()方法获取许可证，用release()归还许可证；tryAcquire()方法尝试获取许可证；

其它方法：

intavailablePermits():返回当前可用许可证数
intgetAQueueLength():返回正在等待获取许可证的线程数
booleanhasQueueThreads():是否有线程正在等待获取许可证
void reducePermits(int reduction):减少reduction个许可证
Collection getQueuedThreads():返回所有等待获取许可证的线程集合
1
2
3
4
5

5.线程间的交互Exchanger

Exchanger用于进行线程间的数据交换。如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange方法；

Exchanger可以用于遗传算法，也可以用于校对工作；

exchange(V x,longtimeout,TimeUnit unit)可以设置最大等待时间

//通过Exchanger交互数据
public class ExchangerTest {
    private static final Exchanger<String> exgr = new Exchanger<>();
    private static final ExecutorService executorService = new ThreadPoolExecutor(
            3, 5,
            10, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(3),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) {
        executorService.execute(()->{
            try{
                String A = "这是A";
               String aGet = exgr.exchange(A);
                System.out.println("aGet:"+aGet);
            }catch (InterruptedException e){
                System.out.println(e);
            }
        });
        executorService.execute(()->{
            String B = "银行流水B";
            try {
                String bGet = exgr.exchange(B);
                System.out.println("bGet:"+bGet);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        executorService.shutdown();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33