JUC

• java.util.concurrent

线程

• 程序

  • 进程
    • 线程
      • Thread.State
        • NEW:新建
        • TUNNABLE:准备就绪
        • BLOCKED:阻塞
        • WAITING:不见不散
        • TIMED_WAITING:过时不候
        • TERMINATED:终结
      • wait
        • Object方法
        • 占用锁，会释放锁
      • sleep
        • Thread方法
        • 不占用锁，不会释放锁
      • 串行
      • 并行
      • 管程
        • 同步机制，保证同一时间只有一个线程访问
      • 用户线程
        • 自定义线程
        • 主线程结束，用户线程还未结束，还可以继续运行
      • 守护线程
        • 比如垃圾回收
        • 主线程结束，守护线程不可以继续运行

• 守护线程

Thread aa = new Thread(()->{System.out.println(Thread.currentThread().getName()+"::"+Thread.currentThread().isDaemon);while(true){}},"aa");
// 设置守护线程
aa.setDaemon(true);
aa.start();
System.out.println(Thread.currentThread().getName() + "over");

lock

• lock

  • 接口，手动上锁解锁

• sunchronized

  • 关键字，自动上锁解锁

• 多线程编程步骤

  • 1、创建资源类，属性和方法
  • 2、在资源类中操作方法
    • 判断
    • 干活
    • 通知
  • 3、创建多个线程，调用资源类方法

sunchronized

// 资源类
public class Ticket{private int num = 30;// 操作方法public synchronized void sale(){// 判断是否有票if(num>0){System.out.println(Thread.currentThread().getName() + "卖出："+(numbr--) + " 剩下：" + num);}}
}

public class SaleTicket{// 创建多个线程，调用资源类方法pubic static void main(String[] args){// 创建TicketTicket ticket = new Ticket();// 创建三个线程new Thread(()->{for(int i=0;i<40;i++){ticket.sale();}},"AA").start();new Thread(()->{for(int i=0;i<40;i++){ticket.sale();}},"BB").start();new Thread(()->{for(int i=0;i<40;i++){ticket.sale();}},"CC").start();}}

lock

• 可重入锁

// 资源类
public class Ticket{private int num = 30;// 创建可重入锁private final ReentrantLock lock = new ReentrantLock();// 操作方法public void sale(){// 上锁lock.lock();try{// 判断是否有票if(num>0){System.out.println(Thread.currentThread().getName() + "卖出："+(numbr--) + " 剩下：" + num);}}finally{// 解锁lock.unlock();}}
}

线程间通讯

synchronized

// 资源类
public class Share{private int num = 0;// +1public synchronized void incr(){// 判断 干活 通知while(num!=0){// 不等于0 等待this.wait();// 在哪里睡，就在哪里醒，所以需要不停判断num值}num++;System.out.println(Thread.currentThread().getName() + ":: " + num);// 通知其他线程,可能虚假唤醒this.notifyAll();}// -1public synchronized void decr(){// 判断 干活 通知while(num!=1){// 不等于1 等待this.wait();}num--;System.out.println(Thread.currentThread().getName() + ":: " + num);// 通知其他线程,可能虚假唤醒this.notifyAll();}
}

public class Demo{// 创建多个线程调用public static void main(String[] args){Share share = new Share();// 创建线程new Thread(()->{for(int i=0;i<10;i++){share.incr();}},"AA").start();new Thread(()->{for(int i=0;i<10;i++){share.decr();}},"BB").start();}}

lock

// 资源类
public class Share{private int num = 0;// 创建lockprivate Lock lock = new ReentrantLock();private Condition condition = lock.newCondition();// +1public synchronized void incr(){// 上锁lock.lock();try{// 判断 干活 通知while(num!=0){// 不等于0 等待condition.await();}num++;System.out.println(Thread.currentThread().getName() + ":: " + num);// 通知其他线程condition.signallAll();}finally{lock.unlock();}}// -1public synchronized void decr(){// 上锁lock.lock();try{// 判断 干活 通知while(num!=0){// 不等于0 等待condition.await();}num--;System.out.println(Thread.currentThread().getName() + ":: " + num);// 通知其他线程condition.signallAll();}finally{lock.unlock();}}
}

定制化通讯

public class ShareResource{// 创建标志位 1-AA 2-BB 3-CCprivate int flag = 1;// 创建lockprivate Lock lock = new ReentrantLock();private Condition c1 = lock.newCondition();private Condition c2 = lock.newCondition();private Condition c3 = lock.newCondition();// 打印5次public void print5(int loop){// 上锁lock.lock();try{// 判断 干活 通知while(flag!=1){// 不等于0 等待c1.await();}for(int i=1;i<=5;i++){System.out.println(Thread.currentThread().getName() + ":: " + i + " 轮数 ："+loop);}// 通知其他线程flag = 2; // 修改标志位c2.signal();}finally{lock.unlock();}}// 打印10次public void print10(int loop){// 上锁lock.lock();try{// 判断 干活 通知while(flag!=2){// 不等于0 等待c2.await();}for(int i=1;i<=10;i++){System.out.println(Thread.currentThread().getName() + ":: " + i + " 轮数 ："+loop);}// 通知其他线程flag = 3; // 修改标志位c3.signal();}finally{lock.unlock();}}// 打印15次public void print10(int loop){// 上锁lock.lock();try{// 判断 干活 通知while(flag!=3){// 不等于0 等待c3.await();}for(int i=1;i<=15;i++){System.out.println(Thread.currentThread().getName() + ":: " + i + " 轮数 ："+loop);}// 通知其他线程flag = 1; // 修改标志位c1.signal();}finally{lock.unlock();}}
}

ShareResource share = new ShareResource();new Thread(()->{for(int i=0;i<10;i++){share.print5(i);} 
},"AA").start();
new Thread(()->{for(int i=0;i<10;i++){share.print10(i);} 
},"BB").start();
new Thread(()->{for(int i=0;i<10;i++){share.print15(i);} 
},"CC").start();

集合线程安全

• 不安全演示

List list = new ArrayList<>();
for(int i=0;i<10;i++){new Thread(()->{// add()不存在synchronized，线程不安全list.add(UUID.randomUUID().toString().substring(0,8));// 获取内容System.out.println(list);// 抛出异常ConcurrentModificationException},String.valueOf(i)).start();
}

Vector

List list = new Vector<>();
for(int i=0;i<10;i++){new Thread(()->{// add()存在synchronized,线程安全list.add(UUID.randomUUID().toString().substring(0,8));// 获取内容System.out.println(list);// 抛出异常ConcurrentModificationException},String.valueOf(i)).start();
}

Collections

• Collections
  • sychronizedList()

List list = Collections.sychronizedList(new ArrayList<>());
for(int i=0;i<10;i++){new Thread(()->{// 线程安全list.add(UUID.randomUUID().toString().substring(0,8));// 获取内容System.out.println(list);// 抛出异常ConcurrentModificationException},String.valueOf(i)).start();
}

CopyOnWriteArrayList

• 线程安全类
  • CopyOnWriteArrayList
  • CopyOnWriteArraySet
  • ConcurrentHashMap

List list = new CopyOnWriteArrayList<>();
for(int i=0;i<10;i++){new Thread(()->{// 线程安全list.add(UUID.randomUUID().toString().substring(0,8));// 获取内容System.out.println(list);// 抛出异常ConcurrentModificationException},String.valueOf(i)).start();
}

多线程锁

• 静态方法
  • 当前类
• 普通方法
  • 当前实例对象
• 同步代码块
  • 配置的对象

public class Phone{public sychronized void sendSMS(){TimeUnit.SECHOUDS.sleep(4);System.out.println("sendSMS");}public static synchronized void sendEmail(){System.out.println("sendEmail");}public void getHello(){System.out.println("getHello");}}

Phone phone = new Phone();new Thread(()->{phone.sendSMS();
},"AA").start();
new Thread(()->{phone.sendEmail();
},"BB").start();

• 公平锁&非公平锁
• 可重入锁
  • synchronized(隐式)
  • lock（显式）

Object o = new Object();
new Thread(()->{synchronized(o){System.out.println(Thread.currentThread().getName()+" 外层");synchronized(o){System.out.println(Thread.currentThread().getName()+" 中层");synchronized(o){System.out.println(Thread.currentThread().getName()+" 内层");}}}
},"t1").star();

Lock o = new ReentrantLock();
new Thread(()->{try{lock.lock();System.out.println(Thread.currentThread().getName()+" 外层");try{lock.lock();System.out.println(Thread.currentThread().getName()+" 中层");try{lock.lock();System.out.println(Thread.currentThread().getName()+" 内层");}finally{lock.unlock();}}finally{lock.unlock();}}finally{lock.unlock();}},"t1").star();

• 死锁
  • jps -l
    • 查看进程
  • jstack xxx
    • 查看进程详情

public class DeadLock{static Object a = new Object();static Object b = new Object();public static main(String[] args){new Thread(()->{sychronized(a){System.out.println(Thread.currentThread().getName()+" 持有a，想要b");TimeUtil.SECONDS.sleep(1);sychronized(b){System.out.println(Thread.currentThread().getName()+" 获取b");}}},"AA").start();new Thread(()->{sychronized(b){System.out.println(Thread.currentThread().getName()+" 持有b，想要a");TimeUtil.SECONDS.sleep(1);sychronized(a){System.out.println(Thread.currentThread().getName()+" 获取a");}}},"BB").start();}
}

callable

• 创建线程方式
  • 继承Thread
  • 实现Runnable接口
    • 没有返回值
    • 没有异常抛出
    • run()
  • 实现Callable接口
    • 有返回值
    • 有异常
    • call()
  • 线程池

public class MyThread1 implements Runnable{public void run(){}
}

public class MyThread2 implements Callable{public Integer call() throws Exception{return 200;}
}

// Runnable
new Thread(new MyThreaad1(),"AA").start();
// Callable
FutureTask futureTask1 = mew FutureTask<>(new MyThread2());
FutureTask futureTask2 = mew FutureTask<>(()->{System.out.println(Thread.currentThread().getName()+" come in callable");return 1024;
});
new Thread(futureTask2,"lucy").start()
while(!futureTask2.isDone()){System.out.println("wait");
}
System.out.println(futureTask2.get());

JUC辅助类

CountDownLatch

public class CountDownLatchDemo{CountDownLatch countDownLatch = new CountDownLatch(6);public static void main(String[] args){// 6 个for(int i=0;i<=6;i++){new Thread(()->{System.out.println(Thread.currentThread().getName());// 计数-1countDownLatch.countDown();},String.valueOf(i)).start();}// 计数为0后放行countDownLatch.await();System.out.println(Thread.currentThread().getName());}
}

CyclicBarrier

public class CyclicBarrierDemo{public static final int NUMBER = 7;public static void main(String[] args){// 7 个CyclicBarrier cyclierBarrier = new CyclierBarrier(NUMBER,()->{System.out.println("已经"+NUMBER+"次了");});for(int i=1;i<=7;i++){new Thread(()->{System.out.println(Thread.currentThread().getName());// 计数+1cyclierBarrier.await();},String.valueOf(i)).start();}}
}

Semaphore

public class SemaphoreDemo{public static void main(String[] args){// 3 个许可数量Semaphore cyclierBarrier = new Semaphore(3);for(int i=1;i<=6;i++){new Thread(()->{try{// 抢占semaphore.acquire();System.out.println(Thread.currentThread().getName()+"抢到");// 设置随机时间TimeUtil.SECONDS.sleep(new Random().nextInt(5));System.out.println(Thread.currentThread().getName()+"离开");}catch(Exception e){}finally{// 释放semaphore.release();}},String.valueOf(i)).start();}}
}

读写锁

• 悲观锁

  • 当前操作人上锁

• 乐观锁

  • 根据版本号进行判断

• 表锁

• 行锁

• 读锁

  • 共享锁
  • 会发生死锁

• 写锁

  • 独占锁
  • 会发生死锁

ReadWriteLock

// 资源类
public class MyCache{private volatile Map map = new HashMap<>();private ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();// 放数据public void put(String key,Object value){// 添加写锁rwLock.writeLock().lock();try{System.out.println(Thread.currentThread().getName()+" 正在操作"+key);TimeUtil.MICROSECONDS.sleep(300);map.put(key,value);System.out.println(Thread.currentThread().getName()+" 写完"+key);}finally{// 释放rwLock.writeLock().unlock();}}public void get(String key){// 添加写锁rwLock.readLock().lock();Object result = null;try{System.out.println(Thread.currentThread().getName()+" 正在读取"+key);TimeUtil.MICROSECONDS.sleep(300);map.put(key,value);result = map.get(key);System.out.println(Thread.currentThread().getName()+" 取完"+key);}finally{// 释放rwLock.readLock().unlock();}return result;}
}

MyCache myCache = new MyCache();
for(int i=1;i<=5;i++){final int num = i;new Thread(()->{myCache.put(num+"",num+"");},String.valueOf(i)).start();
}
for(int i=1;i<=5;i++){final int num = i;new Thread(()->{myCache.get(num+"");},String.valueOf(i)).start();
}

读写锁降级

ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
ReentrantReadWriteLock.ReadLock readLock = rwLock.readLock();// 读锁
ReentrantReadWriteLock.WriteLock writeLock = rwLock.writeLock();// 写锁// 锁降级
// 写锁可降级为读锁
// 读锁不能升级为写锁
// 1 获取写锁
writeLock.lock();
System.out.println("获取写锁");
// 2 获取读锁
readLock.lock();
System.out.println("获取读锁");
writeLock.unlock();
readLock.unlock();

阻塞队列

• BlockingQueue
  • ArrayBlockingQueue
  • LinkedBlockingQueue
  • DelayQueue
  • PriorityBlockingQueue
  • SynchronousQueue
  • LinkedTransferBlockingQueue
  • LinkedBlockingDeque

BlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
// 1
blockingQueue.add("a");// 返回true
blockingQueue.add("b");// 返回true
blockingQueue.add("c");// 返回true
blockingQueue.element();// a// blockingQueue.add("w");// 抛出异常
blockingQueue.remove(); // a
blockingQueue.remove(); // b
blockingQueue.remove(); // c
// blockingQueue.remove(); // 抛出异常// 2
blockingQueue.offer("a");// 返回true
blockingQueue.offer("b");// 返回true
blockingQueue.offer("c");// 返回true
blockingQueue.offer("w");// 返回false
blockingQueue.poll(); // a
blockingQueue.poll(); // b
blockingQueue.poll(); // c
blockingQueue.poll(); // null// 3
blockingQueue.put("a"); 
blockingQueue.put("b"); 
blockingQueue.put("c"); 
blockingQueue.put("w");// 阻塞
blockingQueue.take(); // a
blockingQueue.take(); // b
blockingQueue.take(); // c
blockingQueue.take(); // 阻塞// 4
blockingQueue.offer("a"); 
blockingQueue.offer("b"); 
blockingQueue.offer("c"); 
blockingQueue.offer("w",3L,TimeUnit.SECONDS);// 设置阻塞超时时间

线程池

• 一池N线程
• 一池一线程
• 一池可扩容线程池

public class ThreadPoolDemo{public static void main(String[] args){// 一池多线程ExecutrService threadPool1 = Executors.newFixedThreadPool(5);try{// 10个任务for(int i=0;i<=10;i++){threadPool1.execute(()->{System.out.println(Thread.currentThread().getName());});}}finally{// 关闭threadPool1.shutdown();}}
}

public class ThreadPoolDemo{public static void main(String[] args){// 一池一线程ExecutrService threadPool2 = Executors.newSingleThreadPool();try{// 10个任务for(int i=0;i<=10;i++){threadPool2.execute(()->{System.out.println(Thread.currentThread().getName());});}}finally{// 关闭threadPool2.shutdown();}}
}

public class ThreadPoolDemo{public static void main(String[] args){// 一池可扩容线程池ExecutrService threadPool3 = Executors.newCachedThreadPool();try{// 10个任务for(int i=0;i<=20;i++){threadPool3.execute(()->{System.out.println(Thread.currentThread().getName());});}}finally{// 关闭threadPool3.shutdown();}}
}

线程池7个参数

• int corePoolSize
  • 核心（常用）线程数量
• int maximumPoolSize
  • 最大线程数
• Long keepAliveTime & TimeUnit unit
  • 线程存活时间
• BlockingQueue workQueue
  • 阻塞队列
• ThreadFactory threadFactory
  • 线程工厂
• RejectedExecutionHandler handler
  • 拒绝策略
    • AbortPolicy策略（默认）,抛出异常
    • CallerRunsPolicy,退回调用者
    • DiscardOldestPolicy,抛弃队列中等待最久的任务
    • DiscardPolicy,丢弃，不抛异常

自定义线程池

public class ThreadPoolDemo{public static void main(String[] args){ExecutorService threadPool = new ThreadPoolExecutor(2,5,2L,TimeUnit.SECONDS,new ArrayBlockingQueue<>(3),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());}}

Fork&Join框架

public class MyTask extends RecursiveTask{// 拆分 插值不超过10,计算10以内的计算private static final Integer VALUE = 10; private int begin;private int end;private int result;public MyTask(int begin,int end){this.begin = begin;this.end = end;}protected Integer compute(){// 判断 if((end-begin)<=VALUE){for(int i=begin;i<=end;i++){result = result+i;}}else{// 中间值int middle = (begin+end)/2;// 拆左边MyTask task1 = new MyTask(begin,middle);// 拆右边MyTask task2 = new MyTask(middle+1,end);// 拆分task1.fork();task2.fork();// 合并result = task1.join()+task2.join();}return result;}
}

public static void main(String[] args){MyTask task = new MyTask(0,100);// 分支合并池ForkJoinPool forkJoinPool = new ForkJoinPool();ForkJoinPoolTask forkJoinTask = forkJoinPool.submit(myTask);// 合并结果Integer result = forkJoinTask.get();// 关闭池forkJoinPool.shutdown();}

异步回调

• CompletableFuture

// 异步调用 没有返回值
CompletableFuture c1 = CompletableFuture.runAsync(()->{System.out.println(Thread.currentThread().getName());
});
c1.get();// 异步调用 有返回值
CompletableFuture c2 = CompletableFuture.supplyAsync(()->{System.out.println(Thread.currentThread().getName());int a = i/0;// 模拟异常return 1024;
});c2.whenComplete((t,u)->{System.out.println(t); // 方法的返回值System.out.println(u); // 返回异常
}.get();