Java.Util.Concurrent包中的一些接口和类

1.Callable<V> :接口，多线程的一种实现方式，实现类重写方法，重写的call()方法有返回值或者抛出异常，需要配合着FutureTask类(实现了Runnable接口)使用：

public class CallableTest {
    public static void main(String[] args) {
        Test t=new Test();
        FutureTask<Integer> f=new FutureTask<>(t);  //获取结果，与callable配合着使用  
                                                       //FutureTask类的构造函数必须使用Runnable,或者Callable接口的实例
        new Thread(f).start();   //FutureTask也实现了Runnable接口
        Integer i= null;
        try {
            i = f.get();  //FutureTask的方法，获得个、返回的结果
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println(i);

    }
}
class Test implements Callable<Integer>{  //实现Runnable接口，重写call()方法，有返回值
    int sum=0;
    @Override
    public Integer call() throws Exception {
        for (int i = 0; i <=100; i++) {
            sum=sum+i;
        }
        return sum;
    }
}

2.Semaphore类:信号量，直接new对象，semaphore.acquire():获取信号量，如果获取失败如阻塞,semaphore.relase():释放信号量，semaphore.availablePermits()：获取信号量的编号。

 Semaphore s2=new Semaphore(5,true);
 Semaphore s1=new Semaphore(5);  
  //两种构造方法，(int,boolean) int:资源的访问量  boolean:是否是公平竞争
 //公平竞争：等的越久越先执行

try {
    s1.acquire();//请求资源，如果资源不可用该线程阻塞，直到有可用资源为止；  
} catch (InterruptedException e) {
    e.printStackTrace();
}
s1.release();//释放资源

 一个信号量的停车场案例：

public class Parking2 implements Runnable{
    Semaphore semaphore;
    int id;

    public Parking2(Semaphore semaphore, int id) {
        this.semaphore = semaphore;
        this.id = id;
    }

    public static void main(String[] args) {
        Semaphore semaphore=new Semaphore(4); //4个车位
        ExecutorService executorService= Executors.newCachedThreadPool();//一个线程池
        for (int i = 1; i <= 25; i++) {
            executorService.execute(new Parking2(semaphore,i));
        }
        executorService.shutdown();

    }

    @Override
    public void run() {
        try {
            push();
            Thread.sleep(1000);
            pop();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    //停
    public void push() throws InterruptedException {
        if(semaphore.availablePermits()>0){
            System.out.println(this.id+"可以停车！");
        }else{
            System.out.println(this.id+"请等待，没有车位！");
        }
        semaphore.acquire();
        System.out.println(this.id+"停车成功！");
    }
    //取
    public void pop(){
        semaphore.release();
        System.out.println(this.id+"取车成功！");
    }
}

3.ReentrantLock和Condition:

ReentrantLock:可重入排他锁，乐观锁，直接new对象,需要手动的加和释放锁，但比较灵活。

public class ReentrantTest implements Runnable{

    Test t;
    int id;

    public ReentrantTest(Test t, int id) {
        this.t = t;
        this.id = id;
    }

    @Override
    public void run() {
        try {
            t.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        ExecutorService executorService= Executors.newCachedThreadPool();//线程池
        Test t=new Test();
        //保证每一个线程访问同一个线程池，和同一把锁
        for (int i = 1; i <= 30; i++) {
            executorService.execute(new ReentrantTest(t,i));
        }
        executorService.shutdown();//关闭
    }
}
class Test{
    ReentrantLock reentrantLock=new ReentrantLock();//new一个锁

    public void get() throws InterruptedException {
        reentrantLock.lock(); //上锁
        System.out.println(Thread.currentThread().getName()+"_启动");
        Thread.sleep(1000);
        System.out.println(Thread.currentThread().getName()+"_结束");
        //释放锁
        reentrantLock.unlock();
    }
}

Condition:是一个接口，创建Condition对象需要ReentrantLock.newCondition();主要有await():导致当前线程等待,singal()：唤醒一个等待线程。如下代码实例证明：1.await(),signal()只能在当前对象获得锁(ReentrantLock)是被调用；2.当对象调用await()方法后，会交出锁，让其他线程对象争用；。在调用signal()后，锁对象又回到await()调用处，继续执行。

public class ConditionTest {
    static ReentrantLock lock=new ReentrantLock();
    static Condition condition=lock.newCondition();  //condition是一个接口，有await(),singal()方法
    static int count=1;

    public static void main(String[] args) throws InterruptedException {
        Runnable r1= () -> {
            lock.lock();
            System.out.println("r1带锁启动。。。");
            try {
                System.out.println("r1即将被锁住");
                condition.await();   //等待，并且交出锁
                System.out.println("r1又获新生！");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                System.out.println("r1执行完毕");
                lock.unlock();
            }
        };

        Runnable r2= () -> {
            lock.lock();
            System.out.println("r2带锁启动。。。");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("r1即将被唤醒！");
            condition.signal();  //唤醒r1
            System.out.println("r2执行完毕");
            lock.unlock();

        };
        new Thread(r1).start();  //让r1先执行，但条件不足，只能先等待

        new Thread(r2).start();
    }
}

4.BlockingQueue:是一个阻塞队列的超类接口，此接口的实现类有：ArrayBlockingQueue, DelayQueue, LinkedBlockingDeque, LinkedBlockingQueue, PriorityBlockingQueue, SynchronousQueue 。每个类的具体使用可以参考API。以下的实例使用LinkedBlockingQueue做的。

//设计一个阻塞队列
//每个多线程对象都会往queue(为对象共享static)中放元素会涉及:
//      --->如果队列已满就阻塞等待，--->如果队列空就阻塞等待
//本例中一共有三个执行：  Main主线程， 往blockingqueue放元素的线程 从blockingqueue中取元素的线程
         //三者一起执行
public class TestBlockingQueue implements Runnable{
    //属性：
    int id;  //每个对象编号
    static BlockingQueue<String> queue=new LinkedBlockingQueue<>(3);

    //构造：
    public TestBlockingQueue(int id) {
        this.id = id;
    }

    @Override
    public void run() {
        //每个线程元素都会把自己的ID放到队列中
        try {
            queue.put(String.valueOf(this.id));  //put():会抛异常
            System.out.println(this.id+"已被放进队列");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        //定义一个线程池：
        ExecutorService executorService= Executors.newCachedThreadPool();
        for (int i = 0; i < 15; i++) {
            //线程池执行线程对象
            executorService.execute(new TestBlockingQueue(i));
        }
        Thread t=new Thread(()->{
           //run()的方法体
            while (true){
                try {
                    Thread.sleep(100);
                    if(queue.isEmpty()){//静态属性可以类名直接调
                        break;
                    }
                    //拿出队列中的元素
                    String string=TestBlockingQueue.queue.take();//拿出并删除队列头的元素
                    System.out.println(string+"被拿出");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        //让线程池去执行线程对象t
        executorService.execute(t);

        //关闭线程池
        executorService.shutdown();
    }
}

5.CompletionService:是一个接口,创建实现类对象，是将生产新的异步任务和使用已完成任务结果分离开来的服务。有实现类：ExecutorCompletionService,方法：submit(),take(),具体用法如下实例：

//测试： CompletionService = new ExecutorCompletionService(Executor实现类)--->接口，实现类，让线程池来执行任务
//方法： 1.completionService.submit(Callable<V>的实现类)-->提交执行线程对象
   //   2.completionService.take().get()-->检索并删除下一个已完成的任务，get()获得该线程的执行结果
//会阻塞等待
public class CompletionServiceTest implements Callable<String> {
   int id;

    public CompletionServiceTest(int id) {
        this.id = id;
    }

    @Override
    public String call() throws Exception {
        //每个线程打印，并返回自己的编号
        System.out.println(this.id+"_已启动");
        Thread.sleep(100);
        System.out.println(this.id+"_结束");
        return this.id+"_thread,call()方法的返回值";
    }

    public static void main(String[] args) {
        //线程池
        ExecutorService executorService= Executors.newCachedThreadPool();
                                                                              //放的是executor的实现类(线程池对象)
        CompletionService <String>completionService= new ExecutorCompletionService<String>(executorService);
                    //使用ExecutorCompletionService作为执行对象，LinkedBlockingQueue:作为完成队列

        for (int i = 0; i < 15; i++) {
            completionService.submit(new CompletionServiceTest(i));
             //提交执行
        }
        for (int i = 0; i < 15; i++) {
            try {
                try {
                    System.out.println(completionService.take().get());
                       //并打印返回值
                      //take():获取已完成的任务的结果，并从完成队列中删除该任务
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        //线程池关闭
        executorService.shutdown();
    }
}

6.CountDonwLatch:是继承自Object的类，允许一个或一组线程等待其他线程完成的辅助类

经典案例：10个人一起赛跑，得同时出发。

//CountDownLatch是继承自Object的类
//1.构造方法,直接new： countdownlatck=new CountDownLatch(int)
//2.主要方法：countdownlatch.await():等待计数到达0，后开始线程执行，没有就阻塞等待
//           countdownlatch.countDown():计数减一
//一个10一起赛跑的案例： 要求：10个线程一起开始运行，等10个线程都运行结束后输出比赛结束的信息

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CountDownLatchTest {
    //比赛开始倒计时：
    static CountDownLatch start=new CountDownLatch(1);

    //比赛结束倒计时：
    static CountDownLatch end=new CountDownLatch(10);//10名选手，每来一个就减少一个

    public static void main(String[] args) {
        //线程池
        ExecutorService pool= Executors.newFixedThreadPool(10);
        System.out.println("这是一场10人制比赛~~~");
        for (int i = 0; i < 10; i++) {  //创建10个线程
            int id=i+1;
            Runnable runner= () -> {
                System.out.println("运动员："+id+"准备就绪！");
                try {
                    start.await();  //等待开始的指令

                    Thread.sleep(1000); //跑步中

                    System.out.println("运动员"+id+"已到达！");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    end.countDown();  //每跑完一个人就让计数减一
                }
            };
            pool.execute(runner);  //执行10个线程，都阻塞在开始指令之前
        }

        start.countDown();

        try {
            end.await();  //等待比赛结束
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("全场比赛结束！");
        //关闭线程池
        pool.shutdown();
    }
}

7.CyclicBarrier:是一个类，作用是：允许一组线程互相等待到达一个屏障点前的同步辅助，这个屏障在所有等待的线程被释放后可以重新使用，这是该屏障称为循环(CountDownLatch不同之处)。

//一个旅游团的例子
//构造方法：cyclicBarrier=new CyclicBarrier(int)-->int:要等待的线程数
//重要方法：await()-->当所有线程都执行了该方法，就一起继续向前执行，否则就等待
public class CyclicBarrierTest {
    //属性： 三种策略到达5个城市的时间数组： 西安 北京 重庆 成都 杭州
    static int [] bycar={1,2,3,4,5};
    static int [] bybus={2,3,4,5,6};
    static int [] onfoot={3,4,5,6,7};

    static String getNow(){
        //获取现在的时间
        SimpleDateFormat sdf=new SimpleDateFormat("hh:mm:ss");
        return sdf.format(new Date());
    }
    static class Tour implements Runnable{
        CyclicBarrier cyclicBarrier=new CyclicBarrier(3); //三家旅行公司，采用三种celue
        String tourName; //旅行店的名字
        int [] time;

        public Tour(CyclicBarrier cyclicBarrier, String tourName,int [] time) {
            this.cyclicBarrier = cyclicBarrier;
            this.tourName = tourName;
            this.time=time;
        }

        @Override
        public void run() {
            try {
                Thread.sleep(time[0]*1000);
                System.out.println(getNow()+":"+this.tourName+"到达-->西安");
                try {
                    cyclicBarrier.await();   //等待其他线程执行到此处

                    //继续执行：
                    Thread.sleep(time[1]*1000);
                    System.out.println(getNow()+":"+this.tourName+"到达-->北京");
                    cyclicBarrier.await();  //等待其他两个团到达北京，循环使用

                    Thread.sleep(time[2]*1000);
                    System.out.println(getNow()+":"+this.tourName+"到达-->重庆");
                    cyclicBarrier.await();  //等待其他两个团到达北京，循环使用

                    Thread.sleep(time[3]*1000);
                    System.out.println(getNow()+":"+this.tourName+"到达-->成都");
                    cyclicBarrier.await();  //等待其他两个团到达北京，循环使用

                    Thread.sleep(time[4]*1000);
                    System.out.println(getNow()+":"+this.tourName+"到达-->杭州");
                    //cyclicBarrier.await();  //等待其他两个团到达北京，循环使用
                    
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String[] args) {
        //定义线程池
        ExecutorService pool= Executors.newFixedThreadPool(3);

        //定义CyclicBarrier
        CyclicBarrier cyclicBarrier=new CyclicBarrier(3);

        //线程执行线程,分别执行
        pool.execute(new Tour(cyclicBarrier,"T_Walk",onfoot));
        pool.execute(new Tour(cyclicBarrier,"T_Car",bycar));
        pool.execute(new Tour(cyclicBarrier,"T_Bus",bybus));

        //关闭线程池
        pool.shutdown();
    }
}

8.Future<V>:是一个接口，实现类：FutureTask,CompletableFuture等，get()方法将返回V类型的数据。提供方法来检查计算是否完成，等待其完成，并检索计算结果。 结果只能在计算完成后使用方法get进行检索，如有必要，阻塞，直到准备就绪。 取消由cancel方法执行。 提供其他方法来确定任务是否正常完成或被取消。 计算完成后，不能取消计算。 如果您想使用Future ，以便不可撤销，但不提供可用的结果，则可以声明Future<?>表格的类型，并返回null作为基础任务的结果。

9.ScheduledExecutorService:接口:(实现类：ScheduledExecutorThreadPool)，作用可以让一个EexcutorService在给定延迟后执行，或者是定期执行。

//ScheduledExecutorService:接口:(实现类：ScheduledExecutorThreadPool)
//作用可以让一个EexcutorService在给定延迟后执行，或者是定期执行。
// 方法1：pool.ScheduledFuture<?> scheduleAtFixedRate(Runnable var1, long var2, long var4, TimeUnit var6);
//参数含义：command - 要执行的任务
//         initialDelay - 延迟第一次执行的时间
//         period - 连续执行之间的时期
//         unit - initialDelay和period参数的时间单位

//方法2：ScheduledFuture<?> schedule(Runnable command,
//                            long delay,
//                            TimeUnit unit)创建并执行在给定延迟后启用的单次操作。
//参数
//command - 要执行的任务
//delay - 从现在开始延迟执行的时间
//unit - 延时参数的时间单位
public class ScheduledExecutorServiceTest {
    public static void main(String[] args) {
        //创建线程池
        final ScheduledExecutorService pool= Executors.newScheduledThreadPool(2);
                       //创建一个可以在给定延迟后运行，或者定期执行的线程池
        final Runnable runnable=new Runnable() {
            int count=0;
            @Override
            public void run() {
                System.out.println(new Date()+"runnable"+(++count));
            }
        };

        //1秒后运行，每隔2秒运行一次
        ScheduledFuture<?> scheduledFuture = pool.scheduleAtFixedRate(runnable,1,2,SECONDS);
        //2秒后运行，往后每隔5秒运行一次
        ScheduledFuture<?> scheduledFuture1 = pool.scheduleAtFixedRate(runnable, 2, 5, SECONDS);
        //30秒后结束关闭任务，并关闭Scheduled
        pool.schedule(new Runnable() {
            @Override
            public void run() {
                //关闭操作
                scheduledFuture.cancel(true);
                scheduledFuture1.cancel(true);
                //关闭线程池
                pool.shutdown();
            }
        },30,SECONDS);

    }
}

以上内容参考来自Java ApI文档；

以及推荐博客。