JDK并发包（3）

本是古典何须时尚 2022-05-17 00:49 211阅读 0赞

# 1.循环栅栏：CyclicBarrier #

CyclicBarrier是另一种多线程并发控制实用工具。它和CountDownLatch非常类似，它也可以实现线程间的计数等待，而且他的功能比CountDownLatch更加复杂强大。假如将计数器设为10，那么在凑足10个线程后，计数器会归零，然后接着凑下一批10个线程。这就是它循环栅栏的内在含义。

CyclicBarrier可以接受一个参数作为barrierAction。barrierAction就是当计数器一次次数完成后系统会执行的动作。如下构造器：（parties表示计数总数，也就是参与的线程总数。）

public CyclicBarrier(int parties,Runnable barrierAction);

下面演示下CyclicBarrier的使用：

package thread;
    
    import java.util.Random;
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    
    public class CyclicBarrierDemo {
        public static class Soldier implements Runnable {
            private String soldier;
            private final CyclicBarrier cyclic;
    
            public Soldier(CyclicBarrier cyclic, String soldier) {
                this.soldier = soldier;
                this.cyclic = cyclic;
            }
    
            @Override
            public void run() {
                try {
                    //等待所有士兵到齐
                	System.out.println("等待中……");
                    cyclic.await();
                    System.out.println("等待完成……");
                    doWork();
                    //等待所有士兵完成工作
                    cyclic.await();
                    doWorkTwo();
                    cyclic.await();
                    
                } catch (InterruptedException e) {//在等待过程中,线程被中断
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {//表示当前CyclicBarrier已经损坏.系统无法等到所有线程到齐了.
                    e.printStackTrace();
                }
            }
    
            void doWork() {
                try {
                    Thread.sleep(Math.abs(new Random().nextInt() % 10000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(soldier + ":任务完成");
            }
            
            
            void doWorkTwo() {
                try {
                    Thread.sleep(Math.abs(new Random().nextInt() % 10000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(soldier + ":打仗去啊……");
            }
    
        }
    
        public static class BarrierRun implements Runnable {
            boolean flag;
            int N;
    
            public BarrierRun(boolean flag, int N) {
                this.flag = flag;
                this.N = N;
            }
    
         
            @Override
            public void run() {
                if (flag) {
                    System.out.println("司令:[士兵" + N + "个,任务完成!]");
                } else {
                    System.out.println("司令:[士兵" + N + "个,集合完毕!]");
                    flag = true;
                }
            }
        }
    
        public static void main(String[] args) {
            final int N = 10;
            Thread[] allSoldier = new Thread[N];
            boolean flag = false;
            CyclicBarrier cyclic = new CyclicBarrier(N, new BarrierRun(flag, N));
            //设置屏障点,主要为了执行这个方法
            System.out.println("集合队伍! ");
            for (int i = 0; i < N; i++) {
                System.out.println("士兵" + i + "报道! ");
                allSoldier[i] = new Thread(new Soldier(cyclic, "士兵" + i));
                allSoldier[i].start();
            }
        }
    }

CyclicBarrier cyclic = new CyclicBarrier(N, new BarrierRun(flag, N));创建了CyclicBarrier 的实例并且将计数器设为10，每次计数器达到指标时便执行BarrierRun里面的Run方法。当cyclic.await();再次调用后，会进行下一次的次数。最后输出结果如：

集合队伍! 
    士兵0报道! 
    士兵1报道! 
    等待中……
    士兵2报道! 
    等待中……
    士兵3报道! 
    等待中……
    士兵4报道! 
    等待中……
    士兵5报道! 
    等待中……
    士兵6报道! 
    等待中……
    士兵7报道! 
    等待中……
    士兵8报道! 
    等待中……
    士兵9报道! 
    等待中……
    等待中……
    司令:[士兵10个,集合完毕!]
    等待完成……
    等待完成……
    等待完成……
    等待完成……
    等待完成……
    等待完成……
    等待完成……
    等待完成……
    等待完成……
    等待完成……
    士兵2:任务完成
    士兵0:任务完成
    士兵5:任务完成
    士兵1:任务完成
    士兵3:任务完成
    士兵7:任务完成
    士兵8:任务完成
    士兵6:任务完成
    士兵9:任务完成
    士兵4:任务完成
    司令:[士兵10个,任务完成!]
    士兵0:打仗去啊……
    士兵7:打仗去啊……
    士兵9:打仗去啊……
    士兵6:打仗去啊……
    士兵5:打仗去啊……
    士兵1:打仗去啊……
    士兵8:打仗去啊……
    士兵2:打仗去啊……
    士兵4:打仗去啊……
    士兵3:打仗去啊……
    司令:[士兵10个,任务完成!]

CyclicBarrier.await()方法可能会抛出两个异常，一个是InterruptedException。另一个是BrokenBarrierException，一旦遇到这个异常则表示当前的CyclicBarrier已经破损了系统可能没办法等待所有线程到齐。如果在循环体处加上中断代码，会抛出一个中断异常和9个BrokenBarrierException。这个异常可以避免其他9个线程进行永久的无意义的等待。

![70][]

# 2.线程阻塞工具类：LockSupport #

LockSupport可以让线程在任意位置让线程阻塞。和Thread.suspend()相比，它弥补了由于resume()在前发生导致线程无法继续执行的情况。和Object.wait()相比它不需要获取某个对象的锁，也不会抛出InterruptedException异常。

LockSupport的静态方法park()可以阻塞当前线程，类似的还有parkNanos(),parkUntil()等方法他们实现了一个限时的等待。

package thread;
    
    import java.util.concurrent.locks.LockSupport;
    
    public class LockSupportDemo {
        public static Object u = new Object();
        static ChangeObjectThread t1 = new ChangeObjectThread("t1");
        static ChangeObjectThread t2 = new ChangeObjectThread("t2");
        static ChangeObjectThread t3 = new ChangeObjectThread("t3");
        static ChangeObjectThread t4 = new ChangeObjectThread("t4");
    
        public static class ChangeObjectThread extends Thread {
            public ChangeObjectThread(String name) {
                super.setName(name);
            }
    
            public void run() {
                synchronized (u) {
                    System.out.println("线程：" + getName());
                    LockSupport.park();
                }
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
        	t1.start();
            Thread.sleep(1000);
            t2.start();
            Thread.sleep(1000);
        	t3.start();
            Thread.sleep(1000);
            t4.start();
            LockSupport.unpark(t1);
            LockSupport.unpark(t2);
            LockSupport.unpark(t3);
            LockSupport.unpark(t4);
            t1.join();
            t2.join();
            t3.join();
            t4.join();
        }
    }

![70 1][]

在这里我并没有保证unpark和park的顺序，但是运行后会发现它永远不会因为park方法导致线程永久挂起，它可以一直都正常执行结束。

这是因为LockSupport类使用类似信号量的机制，它给每一个线程都准备了一个许可如果许可可用，那么park()会立即返回并且消费这个许可（将许可变为不可用），如果许可不可用就会阻塞。而unpark()使许可变为可用。（但是和信号量不同的是，它不会累加，你不可能有超过一个许可，它永远只有一个。）

这个特点使得：即使unpark()操作在park()之前，它也可以使下一次的park()操作立即返回，这就是上面代码能顺利结束的最主要原因。同时处于park()挂起的线程会给出一个WAITNG状态，并且标注是park()引起的。而suspend()会给出Runnable状态。这使分析问题变得格外方便。

除了有定时阻塞功能，LockSupport.park()还能支持中断影响。它并不会抛出InterruptedExceptioin异常，它会返回并且从Thread.interrupted()等方法获得中断标记。

package thread;
    
    import java.util.concurrent.locks.LockSupport;
    
    public class LockSupportDemo {
        public static Object u = new Object();
        static ChangeObjectThread t1 = new ChangeObjectThread("t1");
        static ChangeObjectThread t2 = new ChangeObjectThread("t2");
        static ChangeObjectThread t3 = new ChangeObjectThread("t3");
        static ChangeObjectThread t4 = new ChangeObjectThread("t4");
    
        public static class ChangeObjectThread extends Thread {
            public ChangeObjectThread(String name) {
                super.setName(name);
            }
    
            public void run() {
                synchronized (u) {
                    System.out.println("线程：" + getName());
                    LockSupport.park();
                    if(Thread.interrupted()) {
                    	System.out.println(this.getName()+"被中断了");
                    }
                }
                System.out.println(this.getName()+"执行结束了");
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
        	t1.start();
            Thread.sleep(1000);
            t2.start();
            Thread.sleep(1000);
        	t3.start();
            Thread.sleep(1000);
            t2.interrupt();
            t4.start();
            LockSupport.unpark(t1);
            LockSupport.unpark(t2);
            LockSupport.unpark(t3);
            LockSupport.unpark(t4);
            t1.join();
            t2.join();
            t3.join();
            t4.join();
        }
    }

![70 2][]

# 3.Exchanger #

Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。这两个线程通过 exchange方法交换数据，如果第一个线程先执行exchange()方法，它会一直等待第二个线程也 执行exchange方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。  
代码如下：

import java.util.concurrent.Exchanger;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /* 
    * @author zzf 
    * @date 2019年1月25日 下午5:41:56 
    */
    public class ExchangerTest {
    	private static final Exchanger<String> exgr = new Exchanger<String>();
    	private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
    
    	public static void main(String[] args) {
    		threadPool.execute(new Runnable() {
    			@Override
    			public void run() {
    				try {
    					String A = "银行流水A";// A录入银行流水数据
    					String B = exgr.exchange(A);
    					System.out.println(B);
    				} catch (InterruptedException e) {
    
    				}
    			}
    		});
    		threadPool.execute(new Runnable() {
    			@Override
    			public void run() {
    				try {
    					String B = "银行流水B";// B录入银行流水数据
    					String A = exgr.exchange(B);
    					System.out.println("A和B数据是否一致：" + A.equals(B) + "，A录入的是：" + A + "，B录入是：" + B);
    				} catch (InterruptedException e) {
    
    				}
    			}
    		});
    		threadPool.shutdown();
    	}
    }

![20190125174729158.png][]

如果两个线程有一个没有执行exchange()方法，则会一直等待，如果担心有特殊情况发 生，避免一直等待，可以使用exchange（V x，longtimeout，TimeUnit unit）设置最大等待时长。

OK~

参考《实战Java高并发程序设计》

[70]: /images/20220517/6fae9588ec8d49619b7195bce1c76614.png
[70 1]: /images/20220517/f5ba369d395a4e3e93bc27a458d115a8.png
[70 2]: /images/20220517/1b42e9f2cd1941a78f5cb50fa75eb2cf.png
[20190125174729158.png]: /images/20220517/f1ef80711a5b4353861dcb5afc38c1bc.png