java并发包工具（java.util.Concurrent）

一、CyclicBarrier

作用：所有线程准备好才进行，只要一条线程没准备好，都不进行

用法：所有线程准备好以后调用CyclicBarrier的await方法，然后主线程执行CyclicBarrier的countDown方法

实现需求：n个运动员（n个线程），全部准备好了才一起起跑。代码如下

package concurrent019;
import java.io.IOException;  
import java.util.Random;  
import java.util.concurrent.BrokenBarrierException;  
import java.util.concurrent.CyclicBarrier;  
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.Executors; 
public class UseCyclicBarrier {

    static class Runner implements Runnable {  
        private CyclicBarrier barrier;  
        private String name;  
        
        public Runner(CyclicBarrier barrier, String name) {  
            this.barrier = barrier;  
            this.name = name;  
        }  
        @Override  
        public void run() {  
            try {  
                Thread.sleep(1000 * (new Random()).nextInt(5));  
                System.out.println(name + " 准备OK.");  
                barrier.await();  
            } catch (InterruptedException e) {  
                e.printStackTrace();  
            } catch (BrokenBarrierException e) {  
                e.printStackTrace();  
            }  
            System.out.println(name + " Go!!");  
        }  
    } 
    
    public static void main(String[] args) throws IOException, InterruptedException {  
        CyclicBarrier barrier = new CyclicBarrier(3);  // 3 
        ExecutorService executor = Executors.newFixedThreadPool(3);  
        
        executor.submit(new Thread(new Runner(barrier, "zhangsan")));  
        executor.submit(new Thread(new Runner(barrier, "lisi")));  
        executor.submit(new Thread(new Runner(barrier, "wangwu")));  
  
        executor.shutdown();  
    }  
  
}  

 

 

二、CountDownLacth

作用：监听某些初始化操作，等初始化执行完毕以后，通知主线程继续工作（例如zk的初始化）

用法：将需要别的线程来通知的线程调用CountDownLacth的await方法，其他线程执行了之后，允许这个线程执行的时候，执行CountDownLacth的countDown方法

实现需求：启动3个线程，1线程等待2和3线程执行完毕以后再执行，代码如下：

package concurrent019;

import java.util.concurrent.CountDownLatch;

public class UseCountDownLatch {

    public static void main(String[] args) {
        
        final CountDownLatch countDown = new CountDownLatch(2);
        
        Thread t1 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("进入线程t1" + "等待其他线程处理完成...");
                    countDown.await();
                    System.out.println("t1线程继续执行...");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"t1");
        
        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("t2线程进行初始化操作...");
                    Thread.sleep(3000);
                    System.out.println("t2线程初始化完毕，通知t1线程继续...");
                    countDown.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        Thread t3 = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println("t3线程进行初始化操作...");
                    Thread.sleep(4000);
                    System.out.println("t3线程初始化完毕，通知t1线程继续...");
                    countDown.countDown();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        
        t1.start();
        t2.start();
        t3.start();
        
    }
}

 

ps：CyclicBarrier和CountDownLacth的区别

CyclicBarrier是所有线程等待一个线程的信号，CountDownLacth是一个线程等待所有线程的信号

 

三、Callable、Future

Future实现的就是前面讲的Future模式.

实现功能：主线程在执行的过程中，调用了一个或者多个耗时的操作，需要把这个耗时的操作提到另几个线程操作（这几个耗时的操作可以并行执行）

代码实现：

package concurrent019;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

public class UseFuture implements Callable<String>{
    private String para;
    
    public UseFuture(String para){
        this.para = para;
    }
    
    /**
     * 这里是真实的业务逻辑，其执行可能很慢
     */
    @Override
    public String call() throws Exception {
        //模拟执行耗时
        Thread.sleep(3000);
        String result = this.para + "处理完成";
        return result;
    }
    
    //主控制函数
    public static void main(String[] args) throws Exception {
        String queryStr = "query";
        //构造FutureTask，并且传入需要真正进行业务逻辑处理的类,该类一定是实现了Callable接口的类
        FutureTask<String> future = new FutureTask<String>(new UseFuture(queryStr));
        //创建一个固定线程的线程池且线程数为1,
        ExecutorService executor = Executors.newFixedThreadPool(1);
        //这里提交任务future,则开启线程执行RealData的call()方法执行
        executor.submit(future);
        System.out.println("请求完毕");
        try {
            //这里可以做额外的数据操作，也就是主程序执行其他业务逻辑
            Thread.sleep(2000);
        } catch (Exception e) {
            e.printStackTrace();
        }
        //调用获取数据方法,如果call()方法没有执行完成,则依然会进行等待
        System.out.println("数据：" + future.get()); // 其实这块应该开启线程处理
        executor.shutdown();
    }

}

 

四、Semaphore（信号量）

Semaphore是计数信号量。Semaphore管理一系列许可证。每个acquire方法阻塞，直到有一个许可证可以获得然后拿走一个许可证；每个release方法增加一个许可证，这可能会释放一个阻塞的acquire方法。然而，其实并没有实际的许可证这个对象，Semaphore只是维持了一个可获得许可证的数量。（用于限流）

实现需求：总共有20条线程，但是最多只能放5条线程进来执行

代码实例：

package concurrent019;

import java.util.concurrent.ExecutorService;  
import java.util.concurrent.Executors;  
import java.util.concurrent.Semaphore;  
  
public class UseSemaphore {  
  
    public static void main(String[] args) {  
        // 线程池  
        ExecutorService exec = Executors.newCachedThreadPool();  
        // 只能5个线程同时访问  
        final Semaphore semp = new Semaphore(5);  
        // 模拟20个客户端访问  
        for (int index = 0; index < 20; index++) {  
            final int NO = index;  
            Runnable run = new Runnable() {  
                public void run() {  
                    try {  
                        // 获取许可  
                        semp.acquire();  
                        System.out.println("Accessing: " + NO);  
                        //模拟实际业务逻辑
                        Thread.sleep((long) (Math.random() * 10000));  
                        // 访问完后，释放  
                        semp.release();  
                    } catch (InterruptedException e) {  
                    }  
                }  
            };  
            exec.execute(run);  
        } 
        
        try {
            Thread.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        //System.out.println(semp.getQueueLength());
        
        
        
        // 退出线程池  
        exec.shutdown();  
    }  
  
}