java并发工具类：CountDownLatch,CyclicBarrier,Semaphore,Exchange

ゞ浴缸里的玫瑰 2022-05-14 10:06 255阅读 0赞

jdk1.5之后，提供了非常多的辅助类来给我们使用，其中就有CountDownLatch,CyclicBarrier,Semaphore,Exchange

--------------------

一，等待多线程完成CountDownLatch类位于java.util.concurrent包下，利用它可以实现类似计数器的功能。它允许一个或多个线程等待其他线程执行完成操作；功能就是加强的join方法操作；它只有一次赋值，不能重置；

例子:

package countdownlatch;
    
    import java.util.concurrent.CountDownLatch;
    
    /**
     * 类描述:一个线程里面可以分几步进行执行
     * 
     * @author: 张宇
     * @date: 日期: 2018年9月6日 时间: 上午9:20:45
     * @version 1.0
     */
    
    public class CountDownLatchTest {
    	static CountDownLatch c = new CountDownLatch(2);
    
    	public static void main(String[] args) throws InterruptedException {
    		new Thread(new Runnable() {
    			@Override
    			public void run() {
    				// TODO Auto-generated method stub
    				System.out.println("线程1执行任务1");
    				c.countDown();
    				System.out.println("线程2执行任务2");
    				c.countDown();
    			}
    		}).start();
    		c.await();
    		System.out.println("主线程汇入!");
    	}
    }

**执行结果：**

线程1执行任务1
    线程2执行任务2
    主线程汇入!

--------------------

**二，同步屏障CyclicBarrier**

让一组线程到达一个屏障时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续运行。

**例子1:**

package countdownlatch;
    
    import java.util.concurrent.CyclicBarrier;
    /**
    *类描述:测试同步屏障
    *@author: 张宇
    *@date: 日期: 2018年9月6日 时间: 下午2:33:00
    *@version 1.0
     */
    public class CyclicBarrierDemo {
    
    	static CyclicBarrier c = new CyclicBarrier(2);
    
    	public static void main(String[] args) {
    		new Thread(new Runnable() {
    
    			//System.out.println(1);
    			@Override
    			public void run() {
    				// TODO Auto-generated method stub
    				try {
    					c.await();
    				} catch (Exception e) {
    				}
    				System.out.println(1);
    			}
    		}).start();
    
    		try {
    			c.await();
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    		System.out.println(2);
    	}
    }

因为主线程和子线程的调度由cpu决定，两个线程都有可能先执行，所以会产生两种输出

第一种：

1
    2

第二种：

2
    1

如果将new CyclicBarrier（2）改成new CyclicBarrier（3），因为没有第三个await到达屏障，所以之前到达屏障的线程都不会继续运行下去；

**例子2：**

package countdownlatch;
    
    import java.util.concurrent.CyclicBarrier;
    
    /**
    *类描述:测试CyclicBarrierDemo2
    *@author: 张宇
    *@date: 日期: 2018年9月6日 时间: 下午2:46:35
    *@version 1.0
     */
    public class CyclicBarrierDemo2 {
    	static CyclicBarrier c = new CyclicBarrier(2, new A());
    
    	public static void main(String[] args) {
    		new Thread(new Runnable() {
    			@Override
    			public void run() {
    				// TODO Auto-generated method stub
    				try {
    					c.await();
    				} catch (Exception e) {
    				}
    				System.out.println(1);
    			}
    		}).start();
    
    		try {
    			c.await();
    		} catch (Exception e) {
    		}
    		System.out.println(2);
    	}
    
    	static class A implements Runnable {
    		@Override
    		public void run() {
    			// TODO Auto-generated method stub
    			System.out.println(3);
    		}
    	}
    }

**运行结果：**

3
    1
    2

因为CyclicBarrier设置了拦截线程数量为2，所以必须等第一个线程和线程A执行完之后，才会继续执行，最后输出2。

--------------------

CyclicBarrier与CountDownLatch的区别

CountDownLatch的计数器只能使用一次，而CyclicBarrier的计数器可以reset（）方法重置，CyclicBarrier能处理更复杂的业务；

CyclicBarrier还提供其他有用的方法，比如isBroken方法用来解释线程是否被中断。

例子：

package countdownlatch;
    
    import java.util.concurrent.CyclicBarrier;
    
    /**
    *类描述:判断线程是否会中断
    *@author: 张宇
    *@date: 日期: 2018年9月6日 时间: 下午2:56:25
    *@version 1.0
     */
    public class CyclicBarrierTest4 {
    	static CyclicBarrier c = new CyclicBarrier(2);
    
    	public static void main(String[] args) {
    		Thread thread = new Thread(new Runnable() {
    			@Override
    			public void run() {
    				// TODO Auto-generated method stub
    				try {
    					c.await();
    				} catch (Exception e) {
    				}
    			}
    		});
    		thread.start();
    		thread.interrupt();
    
    		try {
    			c.await();
    		} catch (Exception e) {
    			System.out.println(c.isBroken());
    		}
    	}
    }

**运行结果:**

true

--------------------

**三，控制并发线程数的Semaphore  
       **Semaphore是用来控制同时访问特定资源的线程数量，协调各个线程，以保证合理的使用公共资源。

工厂有5台机器，但是有8个工人，一台机器同时只能被一个工人使用，只有使用完了，其他工人才能继续使用。 那么我们就可以通过Semaphore来实现

例子:

package countdownlatch;
    
    import java.util.concurrent.Semaphore;
    
    /**
     * 
     * 
     * @author: 张宇
     * @date: 日期: 2018年9月6日 时间: 上午11:43:18
     * @version 1.0
     */
    public class SemaphoreTest2 {
    	public static void main(String[] args) {
    		int N = 8; // 工人数量
    		Semaphore semaphore = new Semaphore(5);// 机器数量
    		for (int i = 0; i < N; i++) {
    			new Worker(i, semaphore).start();
    		}
    	}
    
    	static class Worker extends Thread {
    		private int num;
    		private Semaphore semaphore;
    
    		public Worker(int num, Semaphore semaphore) {
    			this.num = num;
    			this.semaphore = semaphore;
    		}
    
    		public void run() {
    			try {
    				semaphore.acquire();
    				System.out.println("工人" + this.num + "占用一个机器在生产。。。。");
    				Thread.sleep(2000);
    				System.out.println("工人" + this.num + "释放机器");
    				semaphore.release();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    	}
    }

**执行结果:**

工人1占用一个机器在生产。。。。
    工人6占用一个机器在生产。。。。
    工人4占用一个机器在生产。。。。
    工人2占用一个机器在生产。。。。
    工人0占用一个机器在生产。。。。
    工人4释放机器
    工人0释放机器
    工人3占用一个机器在生产。。。。
    工人1释放机器
    工人5占用一个机器在生产。。。。
    工人6释放机器
    工人2释放机器
    工人7占用一个机器在生产。。。。

Semaphore的用法很简单，首先线程使用它的aquire（）方法获取一个许可证，使用完之后调用release()方法归还许可证。还可以用tryAcquire()方法尝试获取许可证。

--------------------

**四，线程间交互数据的Exchanger**

exchanger可以用于**遗传算法**，也可以用于**校对工作；**如果一个线程执行exchange方法 ，那么它就等待第二个线程执行exchange方法，当两个线程都达到同步点时，这两个线程可以交换数据，将本线程产生的数据传递给对方。

例子：

package countdownlatch;
    
    import java.util.concurrent.Exchanger;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**
    *类描述:在java并发包中的exchange类的使用
    *@author: 张宇
    *@date: 日期: 2018年9月6日 时间: 上午11:16:41
    *@version 1.0
     */
    public class ExchangeTest {
    
    	private static final Exchanger<String> exgr = new Exchanger<>();
    	private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
    
    	public static void main(String[] args) {
    		// 一个线程A
    		threadPool.execute(new Runnable() {
    			@Override
    			public void run() {
    				// TODO Auto-generated method stub
    								
    				try {
    					String A = "银行流水A";
    					String b=exgr.exchange(A);
    					System.out.println("BBBB"+b);
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}
    		});
    
    		// 一个线程B
    		threadPool.execute(new Runnable() {
    			@Override
    			public void run() {
    				// TODO Auto-generated method stub
    				try {
    					String B = "银行流水B";
    					String A = exgr.exchange(B);
    					System.out.println("A:" + A + ";B:" + B + ";A是否等于B:"
    							+ A.equals(B));
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}
    		});
    		
    		threadPool.shutdown();
    	}
    }

**执行结果:**

A:银行流水A;B:银行流水B;A是否等于B:false
    BBBB银行流水B

如果担心特殊情况，避免一直等待，可以使用Exchange（x,time,timeUnit）设置最大等待时间。