Java.Utril.Concurrent
Volatile关键字
避免java虚拟机指令重排序，保证共享数据修改同步，数据可见性。volatile相较于synchronized是一种比较轻量级地同步策略，但不具备互斥性，不能成为synchronized的替代，不能保证原子性。
###示例
1package com.wang.test.juc.test_volatile; 2 3public class TestVolatile { 4 public static void main(String[] args) { 5 TestThread testThread = new TestThread(); 6 7 new Thread(testThread).start(); 8 9 while (true)//在线程运行后，读取线程中的flag值 10 { 11 if(testThread.isFlag()){ 12 System.out.println(" get Flag success! "); 13 break; 14 } 15 } 16 } 17} 18 19class TestThread implements Runnable{ 20 21 private boolean flag = false; 22 23 public boolean isFlag() { 24 return flag; 25 } 26 27 public void setFlag(boolean flag) { 28 this.flag = flag; 29 } 30 31 public void run(){ 32 33 try { 34 Thread.sleep(200); 35 } catch (InterruptedException e) { 36 e.printStackTrace(); 37 } 38 flag = true; 39 System.out.println("flag:" + flag); 40 41 } 42 43} 44
###分析
主线程中的读取操作看似在线程执行后，但并发地执行，在线程更改数据之前，主线程已经读取了数据，共享地数据flag不同步。
###修改
1public class TestVolatile { 2 public static void main(String[] args) { 3 TestThread testThread = new TestThread(); 4 5 new Thread(testThread).start(); 6 7 while (true)//在线程运行后，读取线程中的flag值 8 { 9 if(testThread.isFlag()){ 10 System.out.println(" get Flag success! "); 11 break; 12 } 13 } 14 } 15} 16 17class TestThread implements Runnable{ 18 19 private volatile boolean flag = false; 20 21 public boolean isFlag() { 22 return flag; 23 } 24 25 public void setFlag(boolean flag) { 26 this.flag = flag; 27 } 28 29 public void run(){ 30 try { 31 Thread.sleep(200); 32 } catch (InterruptedException e) { 33 e.printStackTrace(); 34 } 35 flag = true; 36 System.out.println("flag:" + flag); 37 } 38 39}
在Flag属性上添加volatile主方法即可读取正确值。
原子性
###示例
1package com.wang.test.juc.test_volatile; 2 3public class TestAtomic { 4 public static void main(String[] args) { 5 Atomic atomic = new Atomic(); 6 Thread t1 = new Thread(atomic); 7 Thread t2 = new Thread(atomic); 8 t1.start(); 9 t2.start(); 10 } 11 12} 13 14class Atomic implements Runnable{ 15 16 private int i =0; 17 18 public void run() { 19 while(true){ 20 try { 21 Thread.sleep(200); 22 } catch (InterruptedException e) { 23 e.printStackTrace(); 24 } 25System.out.println(Thread.currentThread().getName()+": "+getCountI()); 26 } 27 } 28 29 public int getCountI() 30 { 31 return i++; 32 } 33}
###分析
由于线程中操作i++不具备原子性，线程执行中，数据i的递增会出现重复问题，即i的值不会正常递增，会出现线程做出同样操作的情况。此时因为这个操作不是原子的，使用volitale修饰不能解决同步问题。 
原子类
CAS算法
Compare-And-Swap算法时硬件对于并发操作共享数据的支持，包含三个操作数，内存值V、预估值A、更新值B，只有 V == A 时，V = B执行，否则不进行任何操作。
修改程序
1import java.util.concurrent.atomic.AtomicInteger; 2 3public class TestAtomic { 4 public static void main(String[] args) { 5 Atomic atomic = new Atomic(); 6 Thread t1 = new Thread(atomic); 7 Thread t2 = new Thread(atomic); 8 t1.start(); 9 t2.start(); 10 } 11 12} 13 14class Atomic implements Runnable{ 15 16 17 private AtomicInteger i = new AtomicInteger(0); 18 19 public void run() { 20 while(true){ 21 try { 22 Thread.sleep(200); 23 } catch (InterruptedException e) { 24 e.printStackTrace(); 25 } 26 System.out.println(Thread.currentThread().getName()+": "+getCountI()); 27 } 28 } 29 public int getCountI() 30 { 31 return i.addAndGet(1); 32 } 33}
此时线程中i已经是一个原子类型，那么在对数据进行操作的时候是具备原子性的，所有线程在执行i自增时具有源自性，解决了并发问题。
ConcurrentHashMap
HashTable是线程安全的，在访问HashTable时会加上表锁，将操作转为串行，不允许有空值，效率比较低。
CuncurrentHashMap是线程安全的HashMap，采用锁分段机制，每个数据段都是独立的锁，在访问时，可以并行执行，提高效率
###其他 
ConcurrentSkipListMap：同步的TreeMap
CopyOnWriteArrayList：同步的ArrayList (读取和遍历大于更新)
Collections.synchronizedList(new ArrayList(String))
闭锁
CountDownLatch为同步辅助类，在完成一组正在其他线程中执行的操作之前，允许一个或多个线程一直等待。
###示例
1import java.util.concurrent.CountDownLatch; 2 3public class TestCountDownLatch { 4 public static void main(String[] args) { 5 6 final CountDownLatch countDownLatch = new CountDownLatch(2); 7 //锁值为2 8 9 LatchLock latchLock = new LatchLock(countDownLatch); 10 11 long start = System.currentTimeMillis(); 12 13 Thread t1 = new Thread(latchLock); 14 Thread t2 = new Thread(latchLock); 15 t1.start(); 16 t2.start(); 17 18 try { 19 countDownLatch.await();//锁不为0 主线程等待 20 } catch (InterruptedException e) { 21 e.printStackTrace(); 22 } 23 24 long end = System.currentTimeMillis(); 25 26 System.out.println("Time = [" + (end - start) + "mms"+"]"); 27 } 28} 29 30class LatchLock implements Runnable{ 31 32 private CountDownLatch countDownLatch; 33 public LatchLock(CountDownLatch countDownLatch){ 34 this.countDownLatch = countDownLatch; 35 } 36 public void run() { 37 synchronized (this){ 38 try { 39 for(int i=0;i<1000;i++) 40 { 41 System.out.println(Thread.currentThread().getName()+": "+i); 42 } 43 }finally { 44 countDownLatch.countDown();//线程执行一次锁减一 45 } 46 } 47 } 48}
Callable接口
此接口相较于Runnable接口，可以返回值和抛异常。
###示例
1public class TestCallable { 2 3 public static void main(String[] args) { 4 ThreadCallable testCallable = new ThreadCallable(); 5 6 //执行Callable，需要使用FutureTask 实现类用于接收结果 7 FutureTask<Integer> futureTask = new FutureTask(testCallable); 8 9 Thread thread = new Thread(futureTask); 10 thread.start(); 11 12 Integer result = 0; 13 try { 14 result = futureTask.get(); 15 //此方法将在线程执行结束后才会执行 16 } catch (Exception e) { 17 e.printStackTrace(); 18 } 19 20 System.out.println("result = [" + result + "]"); 21 22 } 23 24} 25 26class ThreadCallable implements Callable<Integer>{ 27 public Integer call() throws Exception { 28 int sum = 0; 29 30 for (int i=0;i<100;i++) { 31 sum+=i; 32 System.out.println("i: "+i); 33 } 34 return sum; 35 } 36}
Lock锁
在解决同步问题时，采用synchronized关键字给代码块加锁或者给方法加锁，关键字加锁方式时隐式的，所的获取和释放由执行过程中的线程自行完成，需要显式地完成加锁和锁释放时，可以使用lock加锁方式。
示例
1package com.wang.test.juc.cchm; 2 3import java.util.concurrent.locks.Lock; 4import java.util.concurrent.locks.ReentrantLock; 5 6public class TestLock { 7 public static void main(String[] args) { 8 9 TestThread testThread = new TestThread(); 10 11 Thread t1 = new Thread(testThread); 12 Thread t2 = new Thread(testThread); 13 Thread t3 = new Thread(testThread); 14 15 t1.start(); 16 t2.start(); 17 t3.start(); 18 } 19} 20 21class TestThread implements Runnable{ 22 23 private Lock lock = new ReentrantLock();//锁 24 25 private int count = 1000; 26 27 public void run() { 28 while (true){ // 自旋等待！ 29 lock.lock(); 30 try { 31 Thread.sleep(1); 32 if (count > 0) 33 System.out.println(Thread.currentThread().getName()+" count :"+ --count); 34 } catch (InterruptedException e) { 35 e.printStackTrace(); 36 }finally { // finally释放锁！ 37 lock.unlock(); 38 } 39 40 } 41 } 42}
等待唤醒
###示例-生产者消费者
1public class TestProductorAndConsumer { 2 public static void main(String[] args) { 3 4 Clerk clerk = new Clerk(); 5 Productor productor = new Productor(clerk); 6 Consumer consumer = new Consumer(clerk); 7 8 new Thread(productor,"Producter").start(); 9 new Thread(consumer,"Customer").start(); 10 11 } 12 13} 14 15class Clerk{ 16 17 private int pruduct = 0; 18 19 public synchronized void income(){ 20 21 if (pruduct >= 10){ 22 System.out.println("Can not add more"); 23 }else{ 24 System.out.println(Thread.currentThread().getName()+": "+ ++pruduct); 25 } 26 } 27 28 public synchronized void sale(){ 29 30 if (pruduct <= 0) {System.out.println("Can not sale anything!");} 31 else { 32 System.out.println(Thread.currentThread().getName()+" :"+ --pruduct); 33 } 34 } 35} 36 37class Productor implements Runnable{ 38 39 private Clerk clerk; 40 41 public Productor(Clerk clerk){ 42 this.clerk = clerk; 43 } 44 45 public void run(){ 46 for (int i=0;i<20;i++){ 47 clerk.income(); 48 } 49 } 50} 51 52class Consumer implements Runnable{ 53 private Clerk clerk; 54 55 public Consumer(Clerk clerk) { 56 this.clerk = clerk; 57 } 58 59 public void run() { 60 for (int i=0;i<20;i++){ 61 clerk.sale(); 62 } 63 } 64}
###分析
生产者消费者都会一直进行，会出现没有产品继续消费和库存已满继续生产，即没有货物依旧被多次消费，无法库存仍旧多次生产。
###改进
1public class TestProductorAndConsumer { 2 public static void main(String[] args) { 3 4 Clerk clerk = new Clerk(); 5 Productor productor = new Productor(clerk); 6 Consumer consumer = new Consumer(clerk); 7 8 new Thread(productor,"Producter").start(); 9 new Thread(consumer,"Customer").start(); 10 11 } 12 13} 14 15class Clerk{ 16 17 private int pruduct = 0; 18 19 public synchronized void income(){ 20 21 if (pruduct >= 10){ 22 System.out.println("Can not add more"); 23 try { 24 this.wait(); 25 } catch (InterruptedException e) { 26 e.printStackTrace(); 27 } 28 }else{ 29 System.out.println(Thread.currentThread().getName()+": "+ ++pruduct); 30 this.notifyAll(); 31 } 32 } 33 34 public synchronized void sale(){ 35 36 if (pruduct <= 0) {System.out.println("Can not sale anything!"); 37 try { 38 this.wait(); 39 } catch (InterruptedException e) { 40 e.printStackTrace(); 41 } 42 } 43 else { 44 System.out.println(Thread.currentThread().getName()+" :"+ --pruduct); 45 this.notifyAll(); 46 } 47 } 48} 49 50class Productor implements Runnable{ 51 52 private Clerk clerk; 53 54 public Productor(Clerk clerk){ 55 this.clerk = clerk; 56 } 57 58 public void run(){ 59 for (int i=0;i<20;i++){ 60 clerk.income(); 61 } 62 } 63} 64 65class Consumer implements Runnable{ 66 private Clerk clerk; 67 68 public Consumer(Clerk clerk) { 69 this.clerk = clerk; 70 } 71 72 public void run() { 73 for (int i=0;i<20;i++){ 74 clerk.sale(); 75 } 76 } 77}
等待唤醒，当发生满货或是销空时，进行等待。以上代码无法结束，最后一次地等待，无法被唤醒，由else引发，继续增加生产者和消费者，将会出现虚假唤醒，必须让它自旋等待。
改进 2.0
1package com.wang.test.juc.cchm; 2 3public class TestProductorAndConsumer { 4 public static void main(String[] args) { 5 6 Clerk clerk = new Clerk(); 7 Productor productor = new Productor(clerk); 8 Consumer consumer = new Consumer(clerk); 9 10 new Thread(productor,"Producter").start(); 11 new Thread(consumer,"Customer").start(); 12 new Thread(productor,"Producter2").start(); 13 new Thread(consumer,"Customer2").start(); 14 15 } 16 17} 18 19class Clerk{ 20 21 private int pruduct = 0; 22 23 public synchronized void income(){ 24 25 while (pruduct >= 10){//自旋 26 System.out.println("Can not add more"); 27 try { 28 this.wait(); 29 } catch (InterruptedException e) { 30 e.printStackTrace(); 31 } 32 } 33 System.out.println(Thread.currentThread().getName()+": "+ ++pruduct); 34 this.notifyAll(); 35 36 } 37 38 public synchronized void sale(){ 39 40 while (pruduct <= 0) {System.out.println("Can not sale anything!"); 41 try { 42 this.wait(); 43 } catch (InterruptedException e) { 44 e.printStackTrace(); 45 } 46 } 47 48 System.out.println(Thread.currentThread().getName()+" :"+ --pruduct); 49 this.notifyAll(); 50 51 } 52} 53 54class Productor implements Runnable{ 55 56 private Clerk clerk; 57 58 public Productor(Clerk clerk){ 59 this.clerk = clerk; 60 } 61 62 public void run(){ 63 for (int i=0;i<20;i++){ 64 clerk.income(); 65 } 66 } 67} 68 69class Consumer implements Runnable{ 70 private Clerk clerk; 71 72 public Consumer(Clerk clerk) { 73 this.clerk = clerk; 74 } 75 76 public void run() { 77 for (int i=0;i<20;i++){ 78 clerk.sale(); 79 } 80 } 81}
同步锁
生产消费模型
1import java.util.concurrent.locks.Condition; 2import java.util.concurrent.locks.Lock; 3import java.util.concurrent.locks.ReentrantLock; 4 5public class TestProductorAndConsumer { 6 public static void main(String[] args) { 7 8 Clerk clerk = new Clerk(); 9 Productor productor = new Productor(clerk); 10 Consumer consumer = new Consumer(clerk); 11 12 new Thread(productor,"Producter").start(); 13 new Thread(consumer,"Customer").start(); 14 new Thread(productor,"Producter2").start(); 15 new Thread(consumer,"Customer2").start(); 16 17 } 18 19} 20 21class Clerk{ 22 23 private int pruduct = 0; 24 25 private Lock lock = new ReentrantLock(); 26 27 private Condition condition = lock.newCondition(); 28 //!!!! 29 30 public void income(){ 31 32 lock.lock(); 33 try { 34 while (pruduct >= 10){ 35 System.out.println("Can not add more"); 36 try { 37 condition.await(); 38 //!!!!! 39 } catch (InterruptedException e) { 40 e.printStackTrace(); 41 } 42 } 43 System.out.println(Thread.currentThread().getName()+": "+ ++pruduct); 44 condition.signalAll(); 45 //!!!! 46 47 }finally { 48 lock.unlock(); 49 } 50 } 51 52 public void sale(){ 53 54 lock.lock(); 55 try { 56 while (pruduct <= 0) {System.out.println("Can not sale anything!"); 57 try { 58 condition.await(); 59 } catch (InterruptedException e) { 60 e.printStackTrace(); 61 } 62 } 63 64 System.out.println(Thread.currentThread().getName()+" :"+ --pruduct); 65 condition.signalAll(); 66 }finally { 67 lock.unlock(); 68 } 69 } 70} 71 72class Productor implements Runnable{ 73 74 private Clerk clerk; 75 76 public Productor(Clerk clerk){ 77 this.clerk = clerk; 78 } 79 80 public void run(){ 81 for (int i=0;i<20;i++){ 82 clerk.income(); 83 } 84 } 85} 86 87class Consumer implements Runnable{ 88 private Clerk clerk; 89 90 public Consumer(Clerk clerk) { 91 this.clerk = clerk; 92 } 93 94 public void run() { 95 for (int i=0;i<20;i++){ 96 clerk.sale(); 97 } 98 } 99}
示例交替打印
1import java.util.concurrent.locks.Condition; 2import java.util.concurrent.locks.Lock; 3import java.util.concurrent.locks.ReentrantLock; 4 5public class TestPrintInOrder { 6 public static void main(String[] args) { 7 final Alternate alternate = new Alternate(); 8 new Thread(new Runnable() { 9 public void run() { 10 11 for (int i=0;i<20;i++){ 12 alternate.PrintA(); 13 } 14 } 15 }).start(); 16 new Thread(new Runnable() { 17 public void run() { 18 19 for (int i=0;i<20;i++){ 20 alternate.PrintB(); 21 } 22 } 23 }).start(); 24 new Thread(new Runnable() { 25 public void run() { 26 27 for (int i=0;i<20;i++){ 28 alternate.PrintC(); 29 } 30 } 31 }).start(); 32 } 33} 34 35class Alternate{ 36 private int mark = 1; 37 38 private Lock lock = new ReentrantLock(); 39 private Condition c1 =lock.newCondition(); 40 private Condition c2 =lock.newCondition(); 41 private Condition c3 =lock.newCondition(); 42 43 public void PrintA(){ 44 lock.lock(); 45 try{ 46 while (mark != 1){ 47 c1.await(); 48 } 49 50 System.out.println(Thread.currentThread().getName()+": "+"A"); 51 52 mark = 2; 53 c2.signal(); 54 } catch (InterruptedException e) { 55 e.printStackTrace(); 56 } finally { 57 58 lock.unlock(); 59 } 60 } 61 62 public void PrintB(){ 63 lock.lock(); 64 try{ 65 while (mark != 2){ 66 c2.await(); 67 } 68 69 System.out.println(Thread.currentThread().getName()+": "+"B"); 70 71 mark = 3; 72 c3.signal(); 73 } catch (InterruptedException e) { 74 e.printStackTrace(); 75 } finally { 76 77 lock.unlock(); 78 } 79 } 80 81 public void PrintC(){ 82 lock.lock(); 83 try{ 84 while (mark != 3){ 85 c3.await(); 86 } 87 88 System.out.println(Thread.currentThread().getName()+": "+"C"); 89 90 mark = 1; 91 c1.signal(); 92 } catch (InterruptedException e) { 93 e.printStackTrace(); 94 } finally { 95 96 lock.unlock(); 97 } 98 } 99}
读写锁
当数据在进行写入时，读取操作需要保持同步，即读写应当时互斥的，读取锁可以共享，写入锁独占。
###示例
1import java.util.concurrent.locks.ReadWriteLock; 2import java.util.concurrent.locks.ReentrantReadWriteLock; 3 4public class TestReadWriteLock { 5 public static void main(String[] args) { 6 final TestLockRW testLockRW = new TestLockRW(); 7 8 new Thread(new Runnable() { 9 public void run() { 10 testLockRW.write((int)(Math.random()*1000)); 11 } 12 }).start(); 13 14 for (int i=0;i<20;i++){ 15 new Thread(new Runnable() { 16 public void run() { 17 testLockRW.read(); 18 } 19 }).start(); 20 } 21 } 22} 23 24class TestLockRW{ 25 26 private int i = 0; 27 28 private ReadWriteLock readWriteLock = new ReentrantReadWriteLock(); 29 30 public void read(){ 31 32 readWriteLock.readLock().lock(); 33 try { 34 System.out.println(Thread.currentThread().getName()+": "+i); 35 }finally { 36 readWriteLock.readLock().unlock(); 37 } 38 39 } 40 41 public void write(int random){ 42 43 readWriteLock.writeLock().lock(); 44 try { 45 System.out.println(Thread.currentThread().getName()+": write "+random); 46 i = random; 47 }finally { 48 readWriteLock.writeLock().unlock(); 49 } 50 } 51}
##线程八锁
1public class TestThread8Monitor { 2 public static void main(String[] args) { 3 Number number = new Number(); 4 Number number2 = new Number(); 5 6 new Thread(new Runnable() { 7 public void run() { 8 number.getOne(); 9 } 10 }).start(); 11 new Thread(new Runnable() { 12 public void run() { 13 number2.getTwo(); 14 } 15 }).start(); 16 17// new Thread(new Runnable() { 18// public void run() { 19// number.getThree(); 20// } 21// }).start(); 22 } 23 24} 25 26class Number{ 27 public static synchronized void getOne(){ 28 try { 29 Thread.sleep(3000); 30 } catch (InterruptedException e) { 31 e.printStackTrace(); 32 } 33 System.out.println("One"); 34 } 35 public static synchronized void getTwo(){ 36 System.out.println("Two"); 37 } 38 39 public void getThree(){ 40 System.out.println("Three"); 41 } 42}
- 非静态方法的锁默认为this
- 静态方法的锁为Class实例
- 在某时刻内，只有一个线程拿到锁。
线程池
类比数据库连接池，创建线程和销毁线程比较浪费资源，建立一个线程池，线程池提供一个线程队列，队列中保存着所有等待状态的线程，在需要使用线程时直接在线程池中获取，使用完毕后，归还给线程池，提高相应速度。
体系结构
1java.util.concurrent.Executor 2 |--ExecutorService 线程池主要接口 3 |--ThreadPoolExecutor 线程池实现类 4 |--ScheduleExecutorService 线程调度接口 5 |--ScheduledThreadPoolExecutor 继承线程池实现调度接口 6 7使用方法： 8 工具类：Executors 9 Executors.ewCachedThreadPool() 数量不固定，动态更改数量 10 Executors.newFixedThreadPool(int) 固定容量 11 Executors.newSingleThreadExecutor() 单个线程线程池 12 返回值类型为ExecurotService 13 14 ScheduledThreadPoolExecutor 线程调度 15 16 17 静态方法。 18 19 20import java.util.concurrent.ExecutorService; 21import java.util.concurrent.Executors; 22 23public class TestThreadPool { 24 25 public static void main(String[] args) { 26 27 ExecutorService executorService = Executors.newCachedThreadPool(); 28 29 ThreadPoolimp threadPoolimp = new ThreadPoolimp(); 30 31 32 for ( int i = 0;i<1000;i++){ 33 34 executorService.submit(threadPoolimp); 35 //支持多种线程初始化参数 Runnable、Callable... 36 } 37 38 executorService.shutdown(); 39 40 //new Thread(new ThreadPoolimp()).start(); 41 } 42 43} 44class ThreadPoolimp implements Runnable{ 45 private int i = 0; 46 47 public void run() { 48 while(true){ 49 System.out.println(Thread.currentThread().getName()+" :"+ ++i); 50 } 51 } 52} 53 54 55 56 57 58public class TestScheduledThreadPool { 59 public static void main(String[] args) throws ExecutionException, InterruptedException { 60 61 ScheduledExecutorService pool = Executors.newScheduledThreadPool(5); 62 63 Future<Integer> future = pool.schedule(new Callable<Integer>() { 64 public Integer call() throws Exception { 65 int i =1; 66 System.out.println(Thread.currentThread().getName()); 67 return i; 68 } 69 },5, TimeUnit.SECONDS);//延迟时间 时间单位 70 71 System.out.println(future.get()); 72 pool.shutdown(); 73 } 74} 75
分支合并框架
在必要的情况下，将一个大人物，进行拆分，拆分成若干的小人物，再将一个个小任务的运算结果进行汇总。
1import java.util.concurrent.ForkJoinPool; 2import java.util.concurrent.RecursiveTask; 3 4public class TestForkJoinPool { 5 public static void main(String[] args) { 6 7 ForkJoinPool pool = new ForkJoinPool(); 8 ForkSun forkSun = new ForkSun(0L, 1000000000L); 9 Long sum = pool.invoke(forkSun); 10 System.out.println("sum = [" + sum + "]"); 11 12 } 13} 14class ForkSun extends RecursiveTask<Long>{ 15 16 private static final long serialVersionUID = 7430797084800536110L; 17 private long start; 18 private long end; 19 20 private static final long max = 10000l; 21 22 public ForkSun(long start, long end) { 23 this.start = start; 24 this.end = end; 25 } 26 27 protected Long compute() { 28 long len = end - start; 29 if(len <= max){ 30 long sum = 0L; 31 for(long i = start;i<=end;i++) 32 { 33 sum+=i; 34 } 35 return sum; 36 }else { 37 long middle = (start + end)/2; 38 ForkSun left = new ForkSun(start,middle); 39 left.fork(); 40 ForkSun right = new ForkSun(middle+1,end); 41 right.fork(); 42 43 return left.join() + right.join(); 44 } 45 46 } 47}