1.什么是JUC?
JUC全称 java.util.concurrent 是在并发编程中很常用的实用工具类
2. volatile 关键字内存可见性
2.1 内存可见性问题,先看下面这段代码
package juc; public class TestVolatile { public static void main(String[] args) { ThreadDemo td = new ThreadDemo(); new Thread(td).start(); while (true){ if(td.isFlag()){ System.out.println("-----------------------------"); break; } } } } class ThreadDemo implements Runnable{ private boolean flag = false; @Override public void run() { try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); } flag = true; System.out.println("flag="+flag); } public boolean isFlag(){ return flag; } }
将上面的代码拿到IDEA去运行,发现控制台只打印输出了flag=true,按照正常的情况,应该将 System.out.println(“—————————–“);,此段代码也执行了才对,为什么这里却没有执行呢?这里涉及到了一个内存可见性问题,原因是此段代码中有两个
线程在执行,一个是主线程Main,一个是子线程,JDK会默认为每一个线程都提供一个缓存,提升效率,这就导致了一个问题,两个线程都拥有一个缓存的flag值,子线程虽然执行了flag = true;但此时修改的flag值只是自己副本的flag值,Main也是读取自己的flag值,
所以导致上述的问题存在。
PS:内存可见性问题是,当多个线程操作共享数据时,彼此不可见。
2.2 如何解决?
2.2.1 synchronized 关键字,同步锁能保证数据的及时更新,能够解决问题,但是这样用会导致线程阻塞,影响效率。
while (true){ synchronized (td) { if (td.isFlag()) { System.out.println("-----------------------------"); break; } } }
2.2.2 volatile 关键字:当多个线程操作共享数据时,可以保证内存中的数据可见,相较于synchronized是一种较为轻量级的同步策略。注意:1.volatile 不具备“互斥性”,2.volatile 不能保证变量的“原子性”
private volatile boolean flag = false;
3.原子性
3.1原子性问题,先看下面这段代码
package juc; public class TestAtomicDemo { public static void main(String[] args) { AtomicDemo ad = new AtomicDemo(); for(int i = 0;i<10;i++){ new Thread(ad).start(); } } } class AtomicDemo implements Runnable{ private int serialNumber = 0; @Override public void run() { try { Thread.sleep(200); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(getSerialNumber()); } public int getSerialNumber(){ return serialNumber++; } }
将上面的代码运行,我们发现有几率会出现,原子性问题,那么为什么会出现此问题呢,我们得研究一下i++的原理,i++的操作实际上分为三个步骤“读-改-写”
int i = 10; i = i ++; int temp = i; i = i + 1; i = temp;
通过上面的分析我们可以得出,即使在serialNumber上修饰volatile关键字,也无法将此问题解决,那么我们要如何解决?
3.2 JUC( java.util.concurrent.atomic ) 提供了原子变量
3.2.1 通过观察里面的类,可以发现,里面的类的变量都是用volatile修饰,保证内存可见性,CAS(compare-and-swap)算法保证数据的原子性,CAS算法时硬件对于并发操作共享数据的支持,CAS包含了三个操作数:内存值V预估值A更新值 ,当且仅当
V==A时,V=B,否则,将不做任何操作
// private int serialNumber = 0;
private AtomicInteger serialNumber = new AtomicInteger(0);
将代码修改为原子变量,即可解决上述的原子性问题
4.ConcurrentHashMap锁分段机制
4.1 Java5.0在java.util.concurrent 包中提供了多种并发容器来改进同步容器的性能
4.2 ConcurrentHashMap同步容器是Java5增加的一个线程安全的哈希表,对与多线程的操作,介于HashMap与HashTable之间。内部采用“锁分段”机制代替Hashtable的独占锁。进而提高性能。
4.3 此包还提供了设计用于多线程上下文中的Collection实现:ConcurrentHashMap、ConcurrentSkipListMap、ConcurrentSkipListSet、CopyOnWriteArrayList和CopyOnWriteArraySet。当期望许多线程访问一个给定collection时, ConcurrentHashMap通
常优于同步的HashMap,ConcurrentSkipListMap通常优于同步的TreeMap.当期望的读数和遍历远远大于列表的更新数时,CopyOnWriteArrList优于同步的ArrayList
5.CountDownLatch闭锁操作
5.1 Java5.0在Java.util.concurrent包中提供了多种并发容器类来改进同步容器的性能
5.2 CountDownLatch 一个同步辅助类,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待。
5.3 闭锁可以延迟线程的进度直到其达到终止状态,闭锁可以用来确保某些活动直到其他活动都完成才继续执行:
5.3.1 确保某个计算在其需要的所有资源都被初始化之后才继续执行
5.3.2 确保某个服务在其依赖的所有其他服务都已经启动之后才启动
5.3.3 等待直到某个操作所有参与者都准备就绪在继续执行
5.4 CountDownLatch:闭锁,在完成某些运算时,只有其他所有的线程的运算全部完成,当前运算才算执行。以下代码是用通过闭锁计算10线程执行的时间
5.5 CountDownLatch演示示例代码:
package juc; import java.util.concurrent.CountDownLatch; public class TestCountDownLatch { public static void main(String[] args) { final CountDownLatch latch = new CountDownLatch(5); LatchDemo ld = new LatchDemo(latch); long start = System.currentTimeMillis(); for(int i = 0;i<5;i++){ new Thread(ld).start(); } try { latch.await(); } catch (InterruptedException e) { e.printStackTrace(); } long end = System.currentTimeMillis(); System.out.println("耗费时间为:"+(end - start)); } } class LatchDemo implements Runnable{ private CountDownLatch latch; public LatchDemo(CountDownLatch latch){ this.latch = latch; } @Override public void run() { synchronized (this) { try { for (int i = 0; i < 50000; i++) { if (i % 2 == 0) { System.out.println(i); } } }finally { latch.countDown(); } } } }
6.实现Callable接口
6.1 创建执行线程的方式三:实现Callble接口。相较于实现Runnable接口的方式,方法可以有返回值,并且可以抛出异常
6.2 Callable演示示例代码:
package juc; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.FutureTask; public class TestCallable { public static void main(String[] args) { ThreadDemo td = new ThreadDemo(); // 1.执行Callable方式,需要FutureTask实现类的支持,用于接收运算结果。
FutureTask<Integer> result = new FutureTask<Integer>(td); new Thread(result).start(); // 2.接收线程运算后的结果
try { Integer sum = result.get(); System.out.println(sum); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } } class ThreadDemo implements Callable<Integer>{ @Override public Integer call() throws Exception { int sum = 0; for(int i = 0; i<=100;i++){ System.out.println(i); sum+=i; } return sum; } }
7.Lock同步锁
7.1 用于解决多线程安全问题的方式,synchronized:隐式锁,同步代码块、同步方法Jdk1.5后:同步锁Lock,是一种显式锁 ,需要通过lock()方式上锁,必须通过unlock方法进行释放锁;
7.2 同步锁演示示例代码:
package juc; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class TestLock { public static void main(String[] args) { Ticket tk = new Ticket(); new Thread(tk,"1 号").start(); new Thread(tk,"2 号").start(); new Thread(tk,"3 号").start(); } } class Ticket implements Runnable{ private int tick = 100; private Lock lock = new ReentrantLock(); @Override public void run() { while (true) { lock.lock(); try { if(tick > 0) { try { Thread.sleep(20); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + "完成售票,余票:" + --tick); } } catch (Exception e) { e.printStackTrace(); }finally { lock.unlock(); } } } }
8.如何使用Lock实现等待唤醒机制
8.1 Lock实现等待唤醒机制演示示例代码:
package juc; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class TestProductorAndConsumer { public static void main(String[] args) { Clerk clerk = new Clerk(); Productor pro = new Productor(clerk); Consumer cus = new Consumer(clerk); new Thread(pro,"生成者 A").start(); new Thread(cus,"消费者 ").start(); } } class Clerk{ private int product = 0; private Lock lock = new ReentrantLock(); private Condition condition = lock.newCondition(); public void get(){ lock.lock(); try { while (product >= 1) { System.out.println("产品已满"); try { //this.wait();
condition.await(); } catch (InterruptedException e) { e.printStackTrace(); } } System.out.println(Thread.currentThread().getName() + ":" + ++product); //this.notifyAll();
condition.signalAll(); }finally { lock.unlock(); } } public synchronized void sale(){ lock.lock(); try { while (product <= 0) { System.out.println("缺货"); try { // this.wait();
condition.await(); } catch (InterruptedException e) { e.printStackTrace(); } } System.out.println(Thread.currentThread().getName() + ":" + --product); //this.notifyAll();
condition.signalAll(); }finally { lock.unlock(); } } } class Productor implements Runnable{ private Clerk clerk; public Productor(Clerk clerk){ this.clerk = clerk; } @Override public void run() { for (int i = 0; i < 20 ;i++){ clerk.get(); } } } class Consumer implements Runnable{ private Clerk clerk; public Consumer(Clerk clerk){ this.clerk = clerk; } @Override public void run() { for (int i = 0; i < 20 ; i++) { clerk.sale(); } } }
9.线程按序交替
9.1 线程按序交替演示示例代码:
package juc; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class TestABCAlternate { public static void main(String[] args) { final AlternateDemo ad = new AlternateDemo(); new Thread(new Runnable() { @Override public void run() { for (int i = 1;i<=20;i++){ ad.loopA(i); } } },"A").start(); new Thread(new Runnable() { @Override public void run() { for (int i = 1;i<=20;i++){ ad.loopB(i); } } },"B").start(); new Thread(new Runnable() { @Override public void run() { for (int i = 1;i<=20;i++){ ad.loopC(i); System.out.println("-----------------------------------"); } } },"C").start(); } } class AlternateDemo{ private int number = 1; private Lock lock = new ReentrantLock(); private Condition condition1 = lock.newCondition(); private Condition condition2 = lock.newCondition(); private Condition condition3 = lock.newCondition(); public void loopA(int totalLoop){ lock.lock(); try { // 1. 判断
if(number != 1){ condition1.await(); } // 2.打印
for(int i = 1;i<=5;i++){ System.out.println(Thread.currentThread().getName() + "\t" + i +"\t"+ totalLoop); } number = 2; condition2.signal(); } catch (Exception e){ e.printStackTrace(); } finally { lock.unlock(); } } public void loopB(int totalLoop){ lock.lock(); try{ // 1.判断
if(number != 2){ condition2.await(); } // 2.打印
for(int i = 1;i<=15;i++){ System.out.println(Thread.currentThread().getName() + "\t" + i +"\t"+ totalLoop); } number = 3; condition3.signal(); }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } } public void loopC(int totalLoop){ lock.lock(); try{ // 1.判断
if(number != 3){ condition3.await(); } // 2.打印
for(int i = 1;i<=20;i++){ System.out.println(Thread.currentThread().getName() + "\t" + i +"\t"+ totalLoop); } number = 1; condition1.signal(); }catch (Exception e){ e.printStackTrace(); }finally { lock.unlock(); } } }
10.ReadWriteLock 读写锁
10.1 ReadWriteLock读写锁演示示例代码:
package juc; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; /** * 1. ReadWriteLock: 读写锁 * 写写/读写 需要“互斥” * 读读不需要互斥 */
public class TestReadWriteLock { public static void main(String[] args) { final ReadWriteLockDemo rw = new ReadWriteLockDemo(); new Thread(new Runnable() { @Override public void run() { rw.set((int)(Math.random() * 101 )); } },"Write:").start(); for(int i = 0;i<100;i++ ){ new Thread(new Runnable() { @Override public void run() { rw.get(); } },"Read:").start(); } } } class ReadWriteLockDemo{ private int number = 0; private ReadWriteLock lock = new ReentrantReadWriteLock(); public void get(){ lock.readLock().lock(); // 上锁
try{ System.out.println(Thread.currentThread().getName() + ":" + number ); }finally { lock.readLock().unlock(); } } public void set(int number){ lock.writeLock().lock(); try{ System.out.println(Thread.currentThread().getName()); this.number = number; }finally { lock.writeLock().unlock(); } } }
11.线程八锁
11.1线程八锁演示示例代码:
package juc; /** * 题目:判断打印的“one” or “two”? * * 1.两个普通同步方法,两个线程,标准打印,打印 // one two * 2.新增Thread.sleep() 给getOne() ,打印 // one two * 3.新增普通方法getThread(),打印 // one two * 4.两个普通同步方法,两个Number对象,打印// two one * 5.修改getOne() 为静态同步方法,打印 // two one * 6.修改两个方法均为静态同步方法,一个Number对象 one two * 7.一个静态同步方法,一个非静态同步方法,两个Number对象 two one * 8.两个静态同步方法,两个Number对象 * * 线程八锁的关键: * ① 非静态方法的锁默认为 this,静态方法的锁对应为Class 实例 * ② 某一个时刻内,只能有一个线程持有锁,无论几个方法 */
public class TestThread8Monitor { public static void main(String[] args) { final Number number = new Number(); new Thread(new Runnable() { @Override public void run() { number.getOne(); } }).start(); new Thread(new Runnable() { @Override public void run() { number.getTwo(); } }).start(); new Thread(new Runnable() { @Override public void run() { number.getThree(); } }).start(); } } class Number{ public synchronized void getOne(){ try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("One"); } public synchronized void getTwo(){ System.out.println("Two"); } public void getThree(){ System.out.println("Three"); } }
12.线程池
12.1 线程池:提供了一个线程队列,队列中保存着所有等待状态的线程,避免了创建与销毁额外开销,提高了响应的速度
12.2 线程池的体系结构:
Java.util.concurrent.Executor: 负责线程的使用与调度的根接口;
|– ExecutorService 子接口: 线程池的主要接口;
|– ThreadPoolExecutor 线程池的实现类;
|– ScheduledExecutorService 子接口:负责线程的调度;
|– ScheduledThreadPoolExecutor:继承ThreadPoolExecutor,实现ScheduledExecutorService;
12.3 工具类:Executors
ExecutorService newFixedThreadPool():创建固定大小的线程池;
ExecutorService newCachedThreadPool():缓存线程池,线程池的数量不固定,可以根据需求自动的更改数量;
ExecutorService newSingleThreadExecutor():创建单个线程池,线程池中只有一个线程;
ScheduledExecutorService newScheduledThreadPool():创建固定大小的线程,可以延迟或定时的执行任务;
12.4 线程池演示示例代码:
package juc; import java.util.concurrent.*; public class TestThreadPool { public static void main(String[] args) throws ExecutionException, InterruptedException { // 1.创建线程池
ExecutorService pool = Executors.newFixedThreadPool(5 ); Future<Integer> future = pool.submit(new Callable<Integer>() { @Override public Integer call() throws Exception { int num = 0; for (int i = 0; i < 100; i++) { num +=i; } return num; } }); System.out.println(future.get()); pool.shutdown(); /*ThreadPoolDemo tpd = new ThreadPoolDemo(); // 2.为线程池中的线程分配任务 for (int i = 0; i < 20; i++) { pool.submit(tpd); } // 3.关闭线程池 pool.shutdown(); */ } } class ThreadPoolDemo implements Runnable{ private int i = 0; @Override public void run() { for (int j = 0; j < 20; j++) { System.out.println(Thread.currentThread().getName() + ":" +j ); } } }
12.5 线程调度演示示例代码:
package juc; import java.util.Random; import java.util.concurrent.*; public class TestScheduledThreadPool { public static void main(String[] args) throws ExecutionException, InterruptedException { ScheduledExecutorService pool = Executors.newScheduledThreadPool(5); for(int i = 0;i < 5;i++) { Future<Integer> result = pool.schedule(new Callable<Integer>() { @Override public Integer call() throws Exception { int num = new Random().nextInt(100); System.out.println(Thread.currentThread().getName() + ":" + num); return num; } }, 3, TimeUnit.SECONDS); System.out.println(result.get()); } pool.shutdown(); } }
13.ForkJoinPool分支/合并框架工作窃取
13.1 分支合并框架演示示例代码:
package juc; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.ForkJoinTask; import java.util.concurrent.RecursiveTask; public class TestForkJoinPool { public static void main(String[] args) { ForkJoinPool pool = new ForkJoinPool(); ForkJoinTask<Long> task = new ForkJoinSumCalculate(0L,100000L); Long sum = pool.invoke(task); System.out.println(sum); } } class ForkJoinSumCalculate extends RecursiveTask<Long>{ private long start; private long end; private static final long THURSHOLD = 1000L; // 临界值
public ForkJoinSumCalculate(long start,long end){ this.start = start; this.end = end; } @Override protected Long compute() { long length = end - start; if(length <= THURSHOLD){ long sum = 0L; for (long i = start; i < end; i++) { sum += i; } return sum; }else { long middle = (start + end) / 2; ForkJoinSumCalculate left = new ForkJoinSumCalculate(start,middle); left.fork(); // 进行拆分,同时压入线程队列
ForkJoinSumCalculate right = new ForkJoinSumCalculate(middle+1,end); right.fork(); // 进行拆分,同时压入线程队列
return left.join() + right.join(); } } }