JUC工具类

Bertha 。 2021-10-14 07:46 393阅读 0赞

## 一、概述 ##

1.在并发编程中，我们会用到并发容器，线程池，线程方法。同时如果能够很好的使用JUC下的一些常用工具，对并发编程事半功倍。

2.常见的工具：Semaphore、CountDownLatch、CyclicBarrier、ExChanger、Phaser

## 二、作用 ##

1. Semaphore

通俗来讲这个工具类的作用就是来控制某一资源同一时间段线程数量的访问。

写个简单的列子：

package com.juc.util;
    
    import java.util.concurrent.Semaphore;
    
    public class Test {
    
    	public static void main(String[] args) {
    		MyResource myResource=new MyResource();
    		Thread t1=new MyThread(myResource,"t1");
    		Thread t2=new MyThread(myResource,"t2");
    		Thread t3=new MyThread(myResource,"t3");
    		Thread t4=new MyThread(myResource,"t4");
    		t1.start();
    		t2.start();
    		t3.start();
    		t4.start();
    	}
    }
    
    
    class MyResource{
    	
    	//开启3个线程通道，意味着同一时间段，只能最大允许3个线程访问里面的资源
    	private static Semaphore semaphore=new Semaphore(3);
    	
    	public void server(String str) throws InterruptedException {
    		while(true) {
    			//占用线程通道
    			semaphore.acquire(2);
    			System.out.println(str.toString()+"正在访问资源");
    			Thread.sleep(3000);
    			//释放线程通道
    			semaphore.release(1);
    			System.out.println("访问资源结束");
    		}
    	}
    	
    }
    
    class MyThread extends Thread{
    	
    	private MyResource myResource;
    	
    	public MyThread(MyResource myResource,String name) {
    		this.myResource=myResource;
    		this.setName(name);
    	}
    	
    	@Override
    	public void run() {
    		try {
    			myResource.server(Thread.currentThread().getName());
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    	
    }

2.CountDownLatch

这个工具类在初创建的时候我们可以给它指定一个数，这个数代表着有多少个线程需要操作这个工具类。一个线程完成了自己的任务，可以调用countDown()方法，调用这个方法之后，计数器减一。在这个方法处await（），会对计数器进行判断，直到减为0后，才继续往下执行，不然一直等到。所以，其实这就是给一批线程，然后让每个线程分别去执行任务，最后进行汇总，所以先执行完的线程也必须等到其他线程执行完之后才能继续往下走。

代码：

package com.juc.util;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.Executor;
    import java.util.concurrent.Executors;
    
    public class Test1 {
    	
    	public static void main(String[] args) throws InterruptedException {
    		Executor executor=Executors.newFixedThreadPool(8);
    		CountDownLatch latch=new CountDownLatch(8);
    		for (int i = 0; i < 8; i++) {
    			executor.execute(new Runnable() {
    				@Override
    				public void run() {
    					System.out.println("执行一次");
    					latch.countDown();
    				}
    				
    			});
    					
    		}
    		latch.await();
    		System.out.println("执行完毕");
    	}
    	
    	
    
    }

3.CyclicBarrier

这个和CountDownLatch有点类似，但是这个是通过计数器加1实现的，并且可以复用，主要作用就是，一组线程在shutdown方法之前进行等待。我们给的值是5，调用一次await方法，加1.当有5个线程都调用了加1.这5个线程就可以一起继续往下执行了。

package com.juc.util;
    
    import java.util.concurrent.CyclicBarrier;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class Test2 {
    
    	public static void main(String[] args) throws InterruptedException {
    		
    		CyclicBarrier barrier = new CyclicBarrier(5);
    
            ExecutorService executor = Executors.newCachedThreadPool();
    
            for (int i = 0; i < 10; i++) {
                final int threadNum = i;
                Thread.sleep(1000);
                executor.execute(() -> {
                    try {
                    	  Thread.sleep(1000);
                          barrier.await();
                       System.out.println(threadNum);
                    } catch (Exception e) {
                    }
                });
            }
            executor.shutdown();
            System.out.println("完毕");
        }
    
    }

4.Exchanger

在两个线程之间交换数据，进行线程间的通信。

package com.juc.util;
    
    import java.util.concurrent.Exchanger;
    import java.util.concurrent.TimeUnit;
    
    public class Test3 {
    
    	  public static void main(String[] args) throws InterruptedException {
    	        Exchanger<Integer> exchanger = new Exchanger<Integer>();
    	        Thread producer =new Producer("producer", exchanger);
    	        Thread consumer =new Consumer("consumer", exchanger);
    	        producer.start();
    	        consumer.start();
    	        TimeUnit.SECONDS.sleep(10);
    	        System.out.println("交换完毕");
    	        System.exit(-1);
    	  }
    	  
    	static class Producer extends Thread {
            private Exchanger<Integer> exchanger;
            private static int data = 0;
            Producer(String name, Exchanger<Integer> exchanger) {
                super(name);
                this.exchanger = exchanger;
            }
    
            @Override
            public void run() {
                for (int i=1; i<5; i++) {
                    try {
                        TimeUnit.SECONDS.sleep(1);
                        data = i;
                        System.out.println(getName()+" 交换前:" + data);
                        data = exchanger.exchange(data);
                        System.out.println(getName()+" 交换后:" + data);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    
        static class Consumer extends Thread {
            private Exchanger<Integer> exchanger;
            private static int data = 0;
            Consumer(String name, Exchanger<Integer> exchanger) {
                super(name);
                this.exchanger = exchanger;
            }
    
            @Override
            public void run() {
                while (true) {
                    data = 0;
                    System.out.println(getName()+" 交换前:" + data);
                    try {
                        TimeUnit.SECONDS.sleep(1);
                        data = exchanger.exchange(data);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(getName()+" 交换后:" + data);
                }
            }
        }
    }

5.Phaser

package com.juc.util;
    
    import java.util.concurrent.Phaser;
    
    public class Test4 {
    	
    	public static void main(String[] args) {
    			MyPhaser phaser = new MyPhaser();
    	        StudentTask[] studentTask = new StudentTask[5];
    	        for (int i = 0; i < studentTask.length; i++) {
    	            studentTask[i] = new StudentTask(phaser);
    	            phaser.register();  //注册一次表示phaser维护的线程个数
    	        }
    	 
    	        Thread[] threads = new Thread[studentTask.length];
    	        for (int i = 0; i < studentTask.length; i++) {
    	            threads[i] = new Thread(studentTask[i], "Student "+i);
    	            threads[i].start();
    	        }
    	 
    	        //等待所有线程执行结束
    	        for (int i = 0; i < studentTask.length; i++) {
    	            try {
    	                threads[i].join();
    	            } catch (InterruptedException e) {
    	                e.printStackTrace();
    	            }
    	        }
    	 
    	        System.out.println("Phaser has finished:"+phaser.isTerminated());
    	}
    
    }
    
    class MyPhaser extends Phaser {
        @Override
        protected boolean onAdvance(int phase, int registeredParties) { //在每个阶段执行完成后回调的方法
     
            switch (phase) {
                case 0:
                    return studentArrived();
                case 1:
                    return finishFirstExercise();
                case 2:
                    return finishSecondExercise();
                case 3:
                    return finishExam();
                default:
                    return true;
            }
     
        }
     
        private boolean studentArrived(){
            System.out.println("学生准备好了,学生人数："+getRegisteredParties());
            return false;
        }
     
        private boolean finishFirstExercise(){
            System.out.println("第一题所有学生做完");
            return false;
        }
     
        private boolean finishSecondExercise(){
            System.out.println("第二题所有学生做完");
            return false;
        }
     
        private boolean finishExam(){
            System.out.println("第三题所有学生做完，结束考试");
            return true;
        }
     
    }
    
    class StudentTask implements Runnable {
    	 
        private Phaser phaser;
     
        public StudentTask(Phaser phaser) {
            this.phaser = phaser;
        }
     
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getName()+"到达考试");
            phaser.arriveAndAwaitAdvance();
     
            System.out.println(Thread.currentThread().getName()+"做第1题完成...");
            phaser.arriveAndAwaitAdvance();
     
            System.out.println(Thread.currentThread().getName()+"做第2题完成...");
            phaser.arriveAndAwaitAdvance();
     
            System.out.println(Thread.currentThread().getName()+"做第3题完成...");
            phaser.arriveAndAwaitAdvance();
        }
    }

5.1首先，第一个循环5次，调用了5次phaser.register(),意味着管理着5个线程。

5.2然后循环创建了5个线程，都分别启动。

5.3在每个线程中都调用了4次phaser.arriveAndAwaitAdvance();

5.4一个线程走到phaser.arriveAndAwaitAdvance();这个方法时计数就会减一，因为register注册了5个，所以没有减掉0点时候，就不会继续往下走了，当另外4个线程也调用了这个方法，减到了0.就一起继续往下执行。这时这个register被重置为5.后面重复着相同的步骤。