Java 并发编程实战 学习分享.ppt

1、Java 并发编程实战 学习分享,王清培,目录（要点）：1.基础知识1.1.synchronized、ReentrantLock（内置锁与显示锁）1.2.内存可见性、volatile、ThreadLocal1.3.同步组件 CountDownLatch、FutureTask、Semaphore、CyclicBarrier2.进阶知识2.1.Executor线程池2.2.BoundQueue有界队列2.3.饱和策略3.原理知识3.1.原子变量3.2.CAS、自旋锁3.3.ABA问题、解决（AtomicStampedReference）3.4.非阻塞的链表算法3.5.AQS4.交流4.1.Doub

2、leCheck（双重检查锁的问题）,1.基础知识1.1.synchronized、ReentrantLock（内置锁与显示锁）1.2.内存可见性、volatile、ThreadLocal1.3.同步组件 CountDownLatch、FutureTask、Semaphore、CyclicBarrier,开始!,1.1.synchronized、 ReentrantLock （内置锁与显示锁）,1.synchronized(内置锁),JVM实现 偏向锁、轻量级锁需要手动打开,偏向锁 jdk1.6,轻量锁 jdk1.6,重量锁 原生,object.markword 保存锁的信息,lockrecor

3、d 线程加锁信息,锁升级： 偏向锁(只有一个thread lock)-轻量级锁(CAS lockrecored-markword)-重量级锁,单个Condition queue 原生,1.1.synchronized、 ReentrantLock （内置锁与显示锁）,1.synchronized(内置锁) demo1 锁的方式：,锁定当前对象中的方法-锁定的是当前对象实例,锁定静态方法-锁定的是当前class对象,public synchronized void method() System.out.println(Thread.currentThread().getId(); ,publi

4、c synchronized static void statisMethod() System.out.println(Thread.currentThread().getId(); ,private Object lock = new Object(); public void method1() synchronized (lock) ,锁定代码块-锁定的是lock对象,1.1.synchronized、 ReentrantLock （内置锁与显示锁）,1.synchronized(内置锁) demo2 简单的Condition queue：,public class Condition

5、Queue implements DemoRun private boolean isOk = false; Override public void runTest() throws ExecutionException, InterruptedException, CustomException, BrokenBarrierException Thread thread1 = new Thread() - synchronized (this) while (!isOk) try System.out.println(String.format(“t1:%s wait“, Thread.c

6、urrentThread().getId(); this.wait(); catch (InterruptedException e) e.printStackTrace(); System.out.println(String.format(“t1:%s 唤醒“, Thread.currentThread().getId(); ); thread1.start(); Thread thread2 = new Thread() - synchronized (this) while (!isOk) try System.out.println(String.format(“t2:%s wait

7、“, Thread.currentThread().getId(); this.wait(); catch (InterruptedException e) e.printStackTrace(); System.out.println(String.format(“t2:%s 唤醒“, Thread.currentThread().getId(); ); thread2.start(); Thread.sleep(2000); synchronized (this) isOk = true; this.notifyAll(); ,t1:12 wait t2:13 wait t2:13 唤醒

8、t1:12 唤醒,1.1.synchronized、Lock（内置锁与显示锁）,2.ReentranLock(显示锁),基于AQS实现,相应Interruption,trylock,公平锁、非公平锁,多个Condition queue,1.1.synchronized、Lock（内置锁与显示锁）,2.ReentranLock(显示锁) demo1 trylock 带有超时lock:,/* * nonFair非公平锁、fair公平锁 */ private ReentrantLock lock = new ReentrantLock(true); private void tryLockTest(

9、) throws InterruptedException Thread thread1 = new Thread() - lock.lock(); System.out.println(“thread1:“ + Thread.currentThread().getId(); try Thread.sleep(20000); catch (InterruptedException e) e.printStackTrace(); ); thread1.start(); Thread thread2 = new Thread() - try if (lock.tryLock(5, TimeUnit

10、.SECONDS) System.out.println(“thread2:“ + Thread.currentThread().getId(); try catch (Exception ignored) finally lock.unlock(); catch (InterruptedException e) e.printStackTrace(); ); thread2.start(); System.out.println(“main:“ + Thread.currentThread().getId(); Thread.sleep(10000); ,thread1:12 main:1,

11、1.1.synchronized、Lock（内置锁与显示锁）,2.ReentranLock(显示锁) demo2 响应Interruption,thread2:java.lang.InterruptedException,/* * nonFair非公平锁、fair公平锁 */ private ReentrantLock lock = new ReentrantLock(true); private void lockInterrupted() Thread thread1 = new Thread() - lock.lock(); try Thread.sleep(10000); catch

12、(InterruptedException e) e.printStackTrace(); ); thread1.start(); Thread thread2 = new Thread() - try lock.lockInterruptibly(); lock.lock(); catch (InterruptedException e) System.out.println(String.format(“thread2:%s“, e); ); thread2.start(); try Thread.sleep(3000); catch (InterruptedException e) e.

13、printStackTrace(); thread2.interrupt(); ,1.1.synchronized、Lock（内置锁与显示锁）,2.ReentranLock(显示锁) demo3 多个ConditionQueue 条件队列,public class ConditionDemo implements DemoRun private Lock lock = new ReentrantLock(); private Condition condition1 = lock.newCondition(); private Condition condition2 = lock.newCo

14、ndition(); private class CustomThread extends Thread public CustomThread(String threadName, Runnable runnable) super(runnable); this.setName(threadName); Override public void runTest() throws ExecutionException, InterruptedException, CustomException, BrokenBarrierException for (int i = 0; i try lock

15、.lock(); condition1.await(); System.out.println(String.format(“t:%s 唤醒“, Thread.currentThread().getName(); catch (InterruptedException e) e.printStackTrace(); finally lock.unlock(); ); thread1.start(); CustomThread thread2 = new CustomThread(String.valueOf(i), () - try lock.lock(); condition2.await(

16、); System.out.println(String.format(“t2:%s 唤醒“, Thread.currentThread().getName(); catch (InterruptedException e) e.printStackTrace(); finally lock.unlock(); ); thread2.start(); Thread.sleep(3000); lock.lock(); condition1.signalAll(); lock.unlock(); Thread.sleep(3000); lock.lock(); condition2.signalA

17、ll(); lock.unlock(); ,t:0 唤醒 t:1 唤醒 t:2 唤醒 t:3 唤醒 t:4 唤醒 t:5 唤醒 t:6 唤醒 t:7 唤醒 t:8 唤醒 t:9 唤醒 t2:0 唤醒 t2:1 唤醒 t2:2 唤醒 t2:3 唤醒 t2:4 唤醒 t2:5 唤醒 t2:6 唤醒 t2:7 唤醒 t2:8 唤醒 t2:9 唤醒,可以方便实现读写锁特性,1.2.内存可见性、volatile、ThreadLocal,4.优先使用Volatile 解决可见性、重排序问题,5.ThreadLocal 封闭进当前线程,3.JMM:Happens-before 原则,1.多核处理器带来的内存

18、可见性问题,2.指令重排序问题,1.程序顺序规则,2.监视器锁规则,3.Volatile变量规则,4.线程启动规则,5.程序结束规则,6.中断规则,7.终结器规则,8.传递性规则,1.2.内存可见性、volatile、ThreadLocal,1. ThreadLocal demo：,public class ThreadLocalDemo implements DemoRun private ThreadLocal threadLocal = new ThreadLocal(); Override public void runTest() for (int i = 0; i Product

19、product = new Product(Thread.currentThread().getId(); this.threadLocal.set(product); System.out.println(“set.threadId:“ + product.getThreadId() + “.“ + Thread.currentThread().getId(); product = this.threadLocal.get(); System.out.println(“get.threadId:“ + product.getThreadId() + “.“ + Thread.currentT

20、hread().getId(); ).start(); ,1.3.同步组件 CountDownLatch、FutureTask、Semaphore、CyclicBarrier,1.CountDownLatch demo1 闭锁：,public class CountDownLatchDemo implements DemoRun private CountDownLatch countDownLatch = new CountDownLatch(3); public void runTest() throws InterruptedException for (int i = 0; i try

21、 System.out.println(Thread.currentThread().getId() + “.begin run.“); Thread.sleep(1000); System.out.println(Thread.currentThread().getId() + “.end run.“); countDownLatch.countDown(); catch (InterruptedException e) e.printStackTrace(); ); thread.start(); countDownLatch.await(); System.out.println(Thr

22、ead.currentThread().getId() + “master thread run.“); System.out.println(Thread.currentThread().getId() + “master thread end.“); ,12.begin run. 13.begin run. 14.begin run. 14.end run. 13.end run. 12.end run. 1master thread run. 1master thread end.,1.3.同步组件 CountDownLatch、FutureTask、Semaphore、CyclicBa

23、rrier,1. FutureTask demo2 Future：,Interface Runnable,Interface Future,Interface RunnableFuture,FutureTask,FutureTask futureTask = new FutureTask() - new Product(Thread.currentThread().getId(); futureTask.run(); System.out.println(futureTask.get();,demo.entities.Product682a0b20,ExecutorService thread

24、Pool = Executors.newCachedThreadPool(); Future future = threadPool.submit() - new Product(Thread.currentThread().getId(); System.out.println(future.get();,demo.entities.Product7ba4f24f,可以基于ExecutorService 使用,1.3.同步组件 CountDownLatch、FutureTask、Semaphore、CyclicBarrier,1. Semaphore demo3 信号量：,public cl

25、ass SemaphoreDemo implements DemoRun Semaphore semaphore = new Semaphore(2); int count = 0; Override public void runTest() throws ExecutionException, InterruptedException for (int i = 1; i try this.semaphore.acquire(); count += 1; System.out.println(Thread.currentThread().getId() + “_“ + count); Thr

26、ead.sleep(3000); this.semaphore.release(); catch (InterruptedException e) e.printStackTrace(); ).start(); ,12_2 13_2 14_3 16_4 15_5 17_6 18_7 19_8,1.3.同步组件 CountDownLatch、FutureTask、Semaphore、CyclicBarrier,1. CyclicBarrier demo4 栅栏：,public class CyclicBarrierDemo implements DemoRun private CyclicBar

27、rier cyclicBarrier = new CyclicBarrier(6); public void runTest() throws InterruptedException for (int i = 0; i try System.out.println(Thread.currentThread().getId() + “await.“); cyclicBarrier.await(); Thread.sleep(finalI * 1000); System.out.println(Thread.currentThread().getId() + “go.“); catch (Int

28、erruptedException e) e.printStackTrace(); catch (BrokenBarrierException e) e.printStackTrace(); ); thread.start(); ,12await. 15await. 16await. 17await. 14await. 13await. 12go. 13go. 14go. 15go. 16go. 17go.,1.基础知识1.1.synchronized、ReentrantLock（内置锁与显示锁）1.2.内存可见性、volatile、ThreadLocal1.3.同步组件 CountDownL

29、atch、FutureTask、Semaphore、CyclicBarrier,总结!,2.进阶知识2.1.Executor线程池2.2.BoundQueue有界队列2.2.饱和策略,开始!,2.1.Executor线程池,1.在什么线程中执行任务？,2.任务按照什么顺序执行（FIFO、LIFO、优先级）？,3.有多少个任务能并发执行？,4.在队列中有多少个任务在等待执行？,5.如果系统由于过载而需要拒绝一个任务，那么应该选择哪一个任务？另外，如何通知应用程序有任务被拒绝？,6.在执行一个任务之前或之后，应该进行哪些动作？,1.任务执行策略,2.1.Executor线程池,2.关键属性,Cor

30、e Pool Size 基本大小,Maximum Pool Size 最大大小,Bound Queue 有界队列,饱和策略,2.1.Executor线程池,3.几种常用的线程池,ExecutorService newCachedExecutor = Executors.newCachedThreadPool(); ExecutorService singleExecutor = Executors.newSingleThreadExecutor(); ExecutorService workExecutor = Executors.newWorkStealingPool(); Executor

31、Service fiexdService = Executors.newFixedThreadPool(2); ScheduledExecutorService scheduleExecutor = Executors.newScheduledThreadPool(10);,2.2.BoundQueue有界队列,1.BoundQueue 有界队列线程池,ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 1, 1000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(

32、1);,注意：SynchronousQueue 同步移交队列,2.3.饱和策略,1.饱和策略,(ThreadPoolExecutor) fiexdService). setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy();,AbortPolicy(终止),CallerRunsPolicy(调用者执行),DiscardOldestPolicy(抛弃最旧 ),DiscardPolicy(抛弃),ExecutorService fiexdService = Executors.newFixedThreadPool(2);,2.

33、进阶知识2.1.Executor线程池2.2.BoundQueue有界队列2.3.饱和策略,总结!,3.原理知识3.1.原子变量3.2.CAS、自旋锁3.3.ABA问题、解决（AtomicStampedReference）3.4.非阻塞的链表算法3.5.AQS,开始!,3.1.原子变量,java.util.concurrent.atomic,AtomicInteger,AtomicLong,AtomicReference,AtomicBoolean,AtomicStampedReference,AtomicMarkableReference,AtomicReferenceFieldUpdate

34、r,3.2.CAS、自选锁,1.现代操作系统提供的功能(compare and swap),2.java中的unsafe final native方法,public final native boolean compareAndSwapObject(Object var1, long var2, Object var4, Object var5); public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5); public final native boolean compa

35、reAndSwapLong(Object var1, long var2, long var4, long var6);,sun.misc,3.2.CAS 、自旋锁,3.2.CAS 、自旋锁 demo AtomicInteger:,ThreadId:13,自旋次数:1 ThreadId:13,自旋次数:2 ThreadId:13,自旋次数:3 12 13,public class CASDemo implements DemoRun private AtomicInteger atomicInteger = new AtomicInteger(); Override public void r

36、unTest() for (int i = 1; i int count = 1; while (!atomicIpareAndSet(0, 1) System.out.println( String.format(“ThreadId:%s,自旋次数:%s“, Thread.currentThread().getId(), count); count += 1; try Thread.sleep(1000); catch (InterruptedException e) e.printStackTrace(); try Thread.sleep(3000); atomicIpareAndSet

37、(1, 0); System.out.println(Thread.currentThread().getId(); return; catch (InterruptedException e) e.printStackTrace(); ).start(); ,3.3.ABA问题、解决（AtomicStampedReference）,A,C,单向链表,D,A,D,A,B,Thread 1,Thread 2,Thread 1,D,3.3.ABA问题、解决（AtomicStampedReference）,public class ABA_AtomicStampedReferenceDemo imp

38、lements DemoRun private AtomicInteger atomicInteger = new AtomicInteger(100); private AtomicStampedReference atomicStampedReference = new AtomicStampedReference(100, 0); Override public void runTest() throws ExecutionException, InterruptedException, CustomException, BrokenBarrierException /*修改atomic

39、Integer值*/ this.atomicIpareAndSet(100, 100); /*修改atomicStampedReference值*/ Integer stamp = this.atomicStampedReference.getStamp(); this.atomicStampedRpareAndSet(100, 100, atomicStampedReference.getStamp(), atomicStampedReference.getStamp() + 1); Thread thread = new Thread() - boolean isOk = this.ato

40、micIpareAndSet(100, 200); if (isOk) System.out.println(“atomic integer ok .“); ); thread.start(); Thread thread2 = new Thread() - boolean isOk = this.atomicStampedRpareAndSet(101, 200, stamp, stamp + 1); if (isOk) System.out.println(“atomic stamped reference ok.“); else System.out.println(“atomic st

41、amped reference fail.“); ); thread2.start(); ,3.3.ABA问题、解决（AtomicStampedReference） demo:,atomic integer ok . atomic stamped reference fail.,3.4.非阻塞的链表算法,哑节点,1,2,尾节点,头节点,1.稳定状态,哑节点,1,2,尾节点,头节点,2.中间插入状态,3,哑节点,1,2,尾节点,头节点,3.插入完成，恢复稳定状态,3,3.5.AQS,维护一个state,双向链表保存Thread,private class sync extends Abstrac

42、tQueuedSynchronizer Override protected boolean isHeldExclusively() return getState() = 1; Override protected boolean tryAcquire(int arg) assert arg = 1; if (compareAndSetState(0, 1) setExclusiveOwnerThread(Thread.currentThread(); return true; return false; Override protected boolean tryRelease(int a

43、rg) assert arg = 1; if (getState() = 0) throw new IllegalMonitorStateException(); setExclusiveOwnerThread(null); setState(0); return true; ,重点：,3.5.AQS,public class CustomLockImpl implements demo.concurrentdemo.CustomLock private volatile sync sync = new sync(); Override public void lock() this.sync

44、.acquire(1); Override public void unLock() this.sync.release(1); Override public boolean tryLock() return this.sync.tryAcquire(1); Override public boolean tryRelease() return this.sync.tryRelease(1); Override public void lockInterruptibly() throws InterruptedException sync.acquireInterruptibly(1); ,