Java处理多线程并发相关类

短命女 2022-05-22 02:22 174阅读 0赞

自从JDK5发布以来，在java.util.concurrent包中提供了一些非常有用的辅助类来帮助我们进行并发编程，下面就介绍一下这些辅助类中的Semaphore、CyclicBarrier、CountDownLatch以及Exchanger的相关用法。

一、Semaphore

Semaphore是计数信号量，是操作系统中的一个概念，经常用于限制获取某种资源的线程数量，在new 这个类的时候需要给这个类传递一个参数permits，这个参数是整数类型，这个参数的意思是同一时间内，最多允许多少个线程同时执行acquire方法和release方法之间的代码，如果方法acquire没有参数则默认是一个许可。例如售票窗口，假如有两个窗口，有三十个用户需要买票，控制并发进行编程，代码如下：

public class SemaphoreTest {
    	
    	class SemaphoreRunnable implements Runnable{
    		private Semaphore semaphore;
    		private int user;
    		public SemaphoreRunnable(Semaphore semaphore, int user) {
    			this.semaphore = semaphore;
    			this.user = user;
    		}
    		@Override
    		public void run() {
    			try {
    				//获取semaphore并且释放
    				semaphore.acquire();
    				System.out.println("用户"+user+"进入窗口买票。。。");
    				Thread.sleep((long)(Math.random()*10000));
    				System.out.println("用户"+user+"已经买票。。。");
    				Thread.sleep((long)(Math.random()*10000));
    				System.out.println("用户"+user+"离开窗口。。。");
    				semaphore.release();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    	private void execute(){
    		final Semaphore semaphore = new Semaphore(2);
    		ExecutorService threadPool = Executors.newCachedThreadPool();
    		for(int i=0; i<20; i++){
    			threadPool.execute(new SemaphoreRunnable(semaphore, i+1));
    		}
    		threadPool.shutdown();
    	}
    	public static void main(String[] args) {
    		SemaphoreTest st = new SemaphoreTest();
    		st.execute();
    	}
    }

充分的模拟了现实售票窗口的售票业务，semaphore.acquire()相当于进入了其中一个窗口，买完票后semaphore.release()相当于买票者离开了售票窗口，这种情况下才允许下一个买票者进入售票窗口。

二、CyclicBarrier

CyclicBarrier直译过来叫做内存屏障，它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续下面的业务。CyclicBarrier默认的构造方法是CyclicBarrier(int parties)，其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。与CountDownLatch不同的是该barrier在释放等待线程后可以重用，所以称它为循环(Cyclic)的屏障(Barrier)。业务场景：假设公司突然开窍了，出钱让程序员出去团建，肯定是等所有人员到达指定地点时，才发车，发车之前可能要拍一个集体照，加入就三个程序员，业务代码如下：

public class CyclicBarrierTest {
    	public static void main(String[] args) {
    		final CyclicBarrier cb = new CyclicBarrier(3,new Runnable() {
    			@Override
    			public void run() {
    				System.out.println("人员已经到齐，开始拍照。。。");
    				try {
    					Thread.sleep((long)(Math.random()*10000));
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    		});
    		ExecutorService threadPool = Executors.newCachedThreadPool();
    		for (int i = 0; i < 3; i++) {
    			final int coder = i+1;
    			Runnable r = new Runnable() {
    				@Override
    				public void run() {
    					try {
    						Thread.sleep((long)(Math.random()*10000));
    						System.out.println(coder+"到达集合地点，当前已有"+(cb.getNumberWaiting()+1)+"到达。。。");
    						cb.await();
    						System.out.println("拍照完毕，开始出发。。。");
    						Thread.sleep((long)(Math.random()*1000));
    						System.out.println("到达游玩地，"+coder+"开始下车。。。");
    					} catch (InterruptedException|BrokenBarrierException e) {
    						e.printStackTrace();
    					}
    				}
    			};
    			threadPool.execute(r);
    		}
    		threadPool.shutdown();
    	}
    }

其中cb.await()相当于发出屏障指令，要等所有程序员到达目的地之后才进行下面的安排。查看CyclicBarrier的源码，有两个构造方法：

/**
         * Creates a new {@code CyclicBarrier} that will trip when the
         * given number of parties (threads) are waiting upon it, and which
         * will execute the given barrier action when the barrier is tripped,
         * performed by the last thread entering the barrier.
         *
         * @param parties the number of threads that must invoke {@link #await}
         *        before the barrier is tripped
         * @param barrierAction the command to execute when the barrier is
         *        tripped, or {@code null} if there is no action
         * @throws IllegalArgumentException if {@code parties} is less than 1
         */
        public CyclicBarrier(int parties, Runnable barrierAction) {
            if (parties <= 0) throw new IllegalArgumentException();
            this.parties = parties;
            this.count = parties;
            this.barrierCommand = barrierAction;
        }
    
        /**
         * Creates a new {@code CyclicBarrier} that will trip when the
         * given number of parties (threads) are waiting upon it, and
         * does not perform a predefined action when the barrier is tripped.
         *
         * @param parties the number of threads that must invoke {@link #await}
         *        before the barrier is tripped
         * @throws IllegalArgumentException if {@code parties} is less than 1
         */
        public CyclicBarrier(int parties) {
            this(parties, null);
        }

CyclicBarrier(int parties)中的parties只是指定了需要等待的线程数，而CyclicBarrier(int parties, Runnable barrierAction)则是追加了一个线程，这个线程可以等parties数目线程都到达触发一些业务处理，如上面那个例子当中的人员到达后拍照留念。

三、CountDownLatch

CountDownLatch可以实现类似计数器的功能，计数器的初始值为指定的线程的数量，每当一个线程完成了自己的任务，计数器的值就会减1，当计数器的值达到了0时，它表示所有的线程都完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。构造器上的计数值实际上就是闭锁需要等待的线程数量，这个值只能被设置一次，而且CountDownLatch没有提供任何机制去重新设置这个值。

业务场景：运动员赛跑，要在同一时刻开始起跑，但是要等最后一个运动员到达终点时才会汇总成绩，假如有8个运动员：

public class PlayerTest {
    	public static void main(String[] args) {
    		CountDownLatch beginLatch = new CountDownLatch(1);
    		CountDownLatch endLatch = new CountDownLatch(8);
    		for (int i = 0; i < 8; i++) {
    			new Thread(new PlayerTest().new Work(i+1, beginLatch, endLatch)).start();
    		}
    		try {
    			System.out.println("准备起跑。。。");
    			beginLatch.countDown();
    			endLatch.await();
    			System.out.println("成绩汇总。。。");
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    	class Work implements Runnable{
    		private int id;
    		private CountDownLatch beginLatch;
    		private CountDownLatch endLatch;
    		public Work(int id, CountDownLatch beginLatch, CountDownLatch endLatch) {
    			super();
    			this.id = id;
    			this.beginLatch = beginLatch;
    			this.endLatch = endLatch;
    		}
    		@Override
    		public void run() {
    			try {
    				beginLatch.await();
    				System.out.println("运动员{"+id+"}到达终点。。。");
    				endLatch.countDown();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    }

四、Exchanger

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

业务场景：我们经常看一些贩毒的警匪片，最经典的台词就是“一手交钱，一手交货”，代码如下：

public class ExchangerTest {
    	public static void main(String[] args) {
    		ExecutorService threadPool = Executors.newCachedThreadPool();
    		final Exchanger<String> ex = new Exchanger<>();
    		threadPool.execute(new Runnable() {
    			@Override
    			public void run() {
    				try {
    					String reStr = ex.exchange("一手交货");
    					System.out.println("吸毒者："+reStr);
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    		});
    		threadPool.execute(new Runnable() {
    			@Override
    			public void run() {
    				try {
    					String reStr = ex.exchange("一手交钱");
    					System.out.println("贩毒者："+reStr);
    				} catch (InterruptedException e) {
    					e.printStackTrace();
    				}
    			}
    		});
    		threadPool.shutdown();
    	}
    }

吸毒者要想获得毒品，就要换取“一手交钱”的信息，贩毒者要想获得金钱，就要换取“一手交货”的信息，当双方到达一个小黑屋的同步点时，就可以交钱收货和交货收钱了！

熟悉了各个并发编程提供的控制类的介绍和用法，在一些没有必要使用重量级锁的业务场景中，就可以应用并发挥它们的作用了。