JUC体系结构

作者: 爱情小傻蛋 | 来源:发表于2019-08-22 16:00 被阅读0次

一、J.U.C体系的主要板块

JUC体系包含的内容很多，这里首先列出将会列举的JUC相关的内容。J.U.C体系的主要大板块包含内容如下图：

J.U.C体系

java.util.concurrent在这个包下面主要是
- Executor线程池
- Collections并发集合
- tools并发工具类。
java.util.concurrent.atomic这个包下面是一些原子操作类，算是并发辅助工具类，基本实现依赖于CAS；
java.util.concurrent.locks这个从名字就可以知道它的作用，就是提供锁。

二、Executor

Executor

三、Collections

Collections

四、tools

tools

Tools也包含了5个部分的知识：Executors、Semaphor、Exchanger、CyclicBarrier、CountDownLatch，其实也就是五个工具类，这5个工具类有神马用途呢，就是我们接下来要将的内容了。

Executors：

常见的方法有(都是静态方法)：

1、创建一个指定大小的线程池，如果超过大小，放入blocken队列中，默认是LinkedBlockingQueue，默认的ThreadFactory为：Executors.defaultThreadFactory()，是一个Executors的一个内部类。

Executors.newFixedThreadPool(int)

内部实现是：

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

2、创建一个指定大小的线程池，如果超过大小，放入blocken队列中,默认是LinkedBlockingQueue，自己指定ThreadFactory，自己写的ThreadFactory，必须implements ThreadFactory，实现方法：newThread(Runnable)。
Executors.newFixedThreadPool(int,ThreadFactory)
内部实现是：

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(),
                                  threadFactory);
}

3、创建线程池长度为1的，也就是只有一个长度的线程池，多余的必须等待，它和调用Executors.newFixedThreadPool(1)得到的结果一样：
Executors.newSingleThreadExecutor()

内部实现是：

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

是不是蛮简单的，就是在变参数，你自己也可以new的。
4、和方法3类似，可以自定义ThreadFactory，这里就不多说了！

5、创建可以进行缓存的线程池，默认缓存60s，数据会放在一个SynchronousQueue上，而不会进入blocken队列中，也就是只要有线程进来就直接进入调度，这个不推荐使用，因为容易出问题，除非用来模拟一些并发的测试：

Executors.newCachedThreadPool();

内部实现为：

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

6、和方法5类似，增加自定义ThreadFactory
7、添加一个Schedule的调度器的线程池，默认只有一个调度：

Executors.newSingleThreadScheduledExecutor();

内部实现为(这里可以看到不是用ThreadPoolExector了，schedule换了一个类，内部实现通过ScheduledThreadPoolExecutor类里面的内部类ScheduledFutureTask来实现的，这个内部类是private，默认是引用不到的哦)：

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1));
}

8、和7一样，增加自己定义的ThreadFactory
9、添加一个schedule的线程池调度器，和newFixedThreadPool有点类似：

Executors.newScheduledThreadPool();
内部代码为：

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

Semaphor信号量

信号量可以干什么呢？

答：根据一些阀值做访问控制。

OK，我们这里模拟一个当多个线程并发一段代码的时候，如何控制其访问速度：

import java.util.Random;
import java.util.concurrent.Semaphore;
 
 
  public class SemaphoreTest {
      private final static Semaphore MAX_SEMA_PHORE = new Semaphore(10);
      public static void main(String []args) {
           for(int i = 0 ; i < 100 ; i++) {
                final int num = i;
                final Random radom = new Random();
                new Thread() {
                     public void run() {
                         boolean acquired = false;
                         try {
                              MAX_SEMA_PHORE.acquire();
                              acquired = true;
                              System.out.println("我是线程：" + num + " 我获得了使用权！" + DateTimeUtil.getDateTime());
                              long time = 1000 * Math.max(1, Math.abs(radom.nextInt() % 10));
                              Thread.sleep(time);
                              System.out.println("我是线程：" + num + " 我执行完了！" + DateTimeUtil.getDateTime());
                         }catch(Exception e) {
                              e.printStackTrace();
                         }finally {
                              if(acquired) {
                                 MAX_SEMA_PHORE.release();
                              }
                         }
                      }
                }.start();
           }
      }
  }

这里是简单模拟并发100个线程去访问一段程序，此时要控制最多同时运行的是10个，用到了这个信号量，运行程序用了一个线程睡眠一个随机的时间来代替，你可以看到后面有线程说自己释放了，就有线程获得了，没释放是获取不到的，内部实现方面，我们暂时不管，暂时知道这样用就OK。

Exchanger

Exchanger是线程之间交互数据，且在并发时候使用，两两交换，交换中不会因为线程多而混乱，发送出去没接收到会一直等，由交互器完成交互过程。


import java.util.concurrent.Exchanger;
 
public class ExchangerTest {
    
    public static void main(String []args) {
        final Exchanger <Integer>exchanger = new Exchanger<Integer>();
        for(int i = 0 ; i < 10 ; i++) {
            final Integer num = i;
            new Thread() {
                public void run() {
                    System.out.println("我是线程：Thread_" + this.getName() + "我的数据是：" + num);
                    try {
                        Integer exchangeNum = exchanger.exchange(num);
                        Thread.sleep(1000);
                        System.out.println("我是线程：Thread_" + this.getName() + "我原先的数据为：" + num + " , 交换后的数据为：" + exchangeNum);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }.start();
        }
    }
}

这里运行你可以看到，如果某个线程和另一个线程传送了数据，它接受到的数据必然是另一个线程传递给他的，中间步骤由Exchanger去控制，其实你可以说，我自己随机取选择，不过中间的算法逻辑就要复杂一些了。

CyclicBarrier

CyclicBarrier是当你在很多环节需要卡住，要多个线程同时在这里都达到后，再向下走，很有用途。

能否举个例子，有点抽象？

答：团队出去旅行，大家一起先达到酒店住宿，然后一起达到游乐的地方游玩，然后一起坐车回家，每次需要点名后确认相关人员均达到，然后LZ一声令下，触发，大伙就疯子般的出发了。

下面的例子也是以旅游的方式来呈现给大家：

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
 
public class BarrierTest {
    
    private static final int THREAD_COUNT = 10;
    
    private final static CyclicBarrier CYCLIC_BARRIER = new CyclicBarrier(THREAD_COUNT  ,
        new Runnable() {
            public void run() {
                System.out.println("======>我是导游，本次点名结束，准备走下一个环节!");
            }
        }
    );
    
    public static void main(String []args) 
            throws InterruptedException, BrokenBarrierException {
        for(int i = 0 ; i < 10 ; i++) {
            new Thread(String.valueOf(i)) {
                public void run() {
                    try {
                        System.out.println("我是线程：" + this.getName() + " 我们达到旅游地点！");
                        CYCLIC_BARRIER.await();
                        System.out.println("我是线程：" + this.getName() + " 我开始骑车！");
                        CYCLIC_BARRIER.await();
                        System.out.println("我是线程：" + this.getName() + " 我们开始爬山！");
                        CYCLIC_BARRIER.await();
                        System.out.println("我是线程：" + this.getName() + " 我们回宾馆休息！");
                        CYCLIC_BARRIER.await();
                        System.out.println("我是线程：" + this.getName() + " 我们开始乘车回家！");
                        CYCLIC_BARRIER.await();
                        System.out.println("我是线程：" + this.getName() + " 我们到家了！");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            }.start();
        }
    }
}

测试结果中可以发现，大家一起走到某个步骤后，导游说：“我是导游，本次点名结束，准备走下一个环节!”，然后才会进入下一个步骤，OK，这个有点意思吧，其实赛马也是这个道理，只是赛马通常只有一个步骤，所以我们还有一个方式是：
CountDownLatch的方式来完成赛马操作，CountDownLatch是用计数器来做的，所以它不可以被复用，如果要多次使用，就要从新new一个出来才可以。我们下面的代码中，用两组赛马，每组5个参与者来，做一个简单测试：

import java.util.concurrent.CountDownLatch;
 
public class CountDownLatchTest {
    
    private final static int GROUP_SIZE = 5;
    
    public static void main(String []args) {
        processOneGroup("分组1");
        processOneGroup("分组2");
    }
    
    private static void processOneGroup(final String groupName) {
        final CountDownLatch start_count_down = new CountDownLatch(1);
        final CountDownLatch end_count_down = new CountDownLatch(GROUP_SIZE);
        System.out.println("==========================>\n分组：" + groupName + "比赛开始：");
        for(int i = 0 ; i < GROUP_SIZE ; i++) {
            new Thread(String.valueOf(i)) {
                public void run() {
                    System.out.println("我是线程组：【" + groupName + "】,第：" + this.getName() + " 号线程,我已经准备就绪！");
                    try {
                        start_count_down.await();//等待开始指令发出即：start_count_down.countDown();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("我是线程组：【" + groupName + "】,第：" + this.getName() + " 号线程,我已执行完成！");
                    end_count_down.countDown();
                }
            }.start();
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("各就各位，预备！");
        start_count_down.countDown();//开始赛跑
        try {
            end_count_down.await();//等待多个赛跑者逐个结束
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("分组：" + groupName + "比赛结束！");
    }
}

有点意思哈，如果你自己用多线程实现是不是有点麻烦，不过你可以用Thread的join方法来实现，也就是线程的发生join的时候，当前线程（一般是主线程）要等到对应线程执行完run方法后才会进入下一步，模拟下：

public class ThreadJoinTest {
    
    private final static int GROUP_SIZE = 5;
 
    public static void main(String []args) throws InterruptedException {
        Thread []threadGroup1 = new Thread[5];
        Thread []threadGroup2 = new Thread[5];
        for(int i = 0 ; i < GROUP_SIZE ; i++) {
            final int num = i;
            threadGroup1[i] = new Thread() {
                public void run() {
                    int j = 0;
                    while(j++ < 10) {
                        System.out.println("我是1号组线程：" + num + " 这个是我第：" + j + " 次运行！");
                    }
                }
            };
            threadGroup2[i] = new Thread() {
                public void run() {
                    int j = 0;
                    while(j++ < 10) {
                        System.out.println("我是2号组线程：" + num + " 这个是我第：" + j + " 次运行！");
                    }
                }
            };
            threadGroup1[i].start();
        }
        for(int i = 0 ; i < GROUP_SIZE ; i++) {
            threadGroup1[i].join();
        }
        System.out.println("-==================>线程组1执行完了，该轮到俺了！");
        for(int i = 0 ; i < GROUP_SIZE ; i++) {
            threadGroup2[i].start();
        }
        for(int i = 0 ; i < GROUP_SIZE ; i++) {
            threadGroup2[i].join();
        }
        System.out.println("全部结束啦！哈哈，回家喝稀饭！");
    }
}

代码是不是繁杂了不少，呵呵，我们再看看上面的信号量，如果不用工具，自己写会咋写，我们模拟CAS锁，使用Atomic配合完成咋来做呢。

import java.util.concurrent.atomic.AtomicInteger;
 
public class ThreadWaitNotify {
    
    private final static int THREAD_COUNT = 100;
    
    private final static int QUERY_MAX_LENGTH = 2;
    
    private final static AtomicInteger NOW_CALL_COUNT = new AtomicInteger(0);
 
    public static void main(String []args) throws InterruptedException {
        Thread []threads = new Thread[THREAD_COUNT];
        for(int i = 0 ; i < THREAD_COUNT ; i++) {
            threads[i] = new Thread(String.valueOf(i)) {
                synchronized public void run() {
                    int nowValue = NOW_CALL_COUNT.get();
                    while(true) {
                        if(nowValue < QUERY_MAX_LENGTH && NOW_CALL_COUNT.compareAndSet(nowValue, nowValue + 1)) {
                            break;//获取到了
                        }
                        try {
                            this.wait(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        nowValue = NOW_CALL_COUNT.get();//获取一个数据，用于对比
                    }
                    System.out.println(this.getName() + "======我开始做操作了！");
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(this.getName() + "======操作结束了！");
                    NOW_CALL_COUNT.getAndDecrement();
                    this.notify();
                }
            };
        }
        for(int i = 0 ; i < THREAD_COUNT ; i++) {
            threads[i].start();
        }
    }
}

五、atomic

atomic

六、locks

locks

JUC体系结构

一、J.U.C体系的主要板块

二、Executor

三、Collections

四、tools

Executors：

Semaphor信号量

Exchanger

CyclicBarrier

五、atomic

六、locks

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读