并发-5CAS与AQS

juc： java.util.concurrent

锁：

　　悲观锁：写的比较多，对数据的增删改，读（查）少。Lock

　　乐观锁：反之，读多写少。版本 

并发编程之 CAS 的原理

什么是CAS

CAS （compareAndSwap），中文叫比较交换，一种无锁原子算法乐观锁。过程是这样：它包含 3 个参数 CAS（V，E，N），V表示要更新变量的值，E表示预期值，N表示新值。仅当 V值等于E值时，才会将V的值设为N，如果V值和E值不同，则说明已经有其他线程做两个更新，则当前线程则什么都不做。最后，CAS 返回当前V的真实值。CAS 操作时抱着乐观的态度进行的，它总是认为自己可以成功完成操作。

CAS的全称为Compare And Swap，直译就是比较交换。是一条CPU的原子指令，其作用是让CPU先进行比较两个值是否相等，然后原子地更新某个位置的值，其实现方式是基于硬件平台的汇编指令，在intel的CPU中，使用的是cmpxchg指令，就是说CAS是靠硬件实现的，从而在硬件层面提升效率。

当多个线程同时使用CAS 操作一个变量时，只有一个会胜出，并成功更新，其余均会失败。失败的线程不会挂起，仅是被告知失败，并且允许再次尝试，当然也允许实现的线程放弃操作。基于这样的原理，CAS 操作即使没有锁，也可以发现其他线程对当前线程的干扰。

与锁相比，使用CAS会使程序看起来更加复杂一些，但由于其非阻塞的，它对死锁问题天生免疫，并且，线程间的相互影响也非常小。更为重要的是，使用无锁的方式完全没有锁竞争带来的系统开销，也没有线程间频繁调度带来的开销，因此，他要比基于锁的方式拥有更优越的性能。

简单的说，CAS 需要你额外给出一个期望值，也就是你认为这个变量现在应该是什么样子的。如果变量不是你想象的那样，哪说明它已经被别人修改过了。你就需要重新读取，再次尝试修改就好了。

优缺点：实现复杂，无锁，不存在阻塞，提高效率，性能好，CPU的吞吐量提高

import java.util.concurrent.atomic.AtomicInteger;

public class CASDome {
    private static volatile  int i=0;
    private static AtomicInteger atomicInteger = new AtomicInteger();
    public static void increase1(){
        i++;
    }

    public static void increase2(){
        atomicInteger.incrementAndGet();
    }

    public static void main(String args[]) throws InterruptedException {
        Thread [] threads1 = new Thread[10];
        for (int i = 0; i < 10; i++) {
           threads1[i] = new Thread(()->{
                CASDome.increase1();
            });
            threads1[i].start();
//join方法的加入，形成ThreadGroup，执行到这里后会将main方法停止，而去执行这个当前线程：API文档解释：等待该线程终止。
//此时主线程与用户线程有了交互性。
            threads1[i].join();
        }
        System.out.println("i:"+i);

        Thread [] threads2 = new Thread[10];
        for (int i = 0; i < 10; i++) {
            threads2[i] = new Thread(()->{
                CASDome.increase2();
            });
            threads2[i].start();
            threads2[i].join();
        }
        System.out.println("atomicInteger:"+atomicInteger);
    }


}　

　

CAS底层原理

这样归功于硬件指令集的发展，实际上，我们可以使用同步将这两个操作变成原子的，但是这么做就没有意义了。所以我们只能靠硬件来完成，硬件保证一个从语义上看起来需要多次操作的行为只通过一条处理器指令就能完成。这类指令常用的有：

1. 测试并设置（Tetst-and-Set）

2. 获取并增加（Fetch-and-Increment）

3. 交换（Swap）

4. 比较并交换（Compare-and-Swap）

5. 加载链接/条件存储（Load-Linked/Store-Conditional）

CPU 实现原子指令有2种方式：

1. 通过总线锁定来保证原子性。

总线锁定其实就是处理器使用了总线锁，所谓总线锁就是使用处理器提供的一个 LOCK# 信号，当一个处理器在总线上输出此信号时，其他处理器的请求将被阻塞住，那么该处理器可以独占共享内存。但是该方法成本太大。因此有了下面的方式。

2、通过缓存锁定来保证原子性。

所谓 缓存锁定 是指内存区域如果被缓存在处理器的缓存行中，并且在Lock 操作期间被锁定，那么当他执行锁操作写回到内存时，处理器不在总线上声言 LOCK# 信号，而时修改内部的内存地址，并允许他的缓存一致性机制来保证操作的原子性，因为缓存一致性机制会阻止同时修改两个以上处理器缓存的内存区域数据（这里和 volatile 的可见性原理相同），当其他处理器回写已被锁定的缓存行的数据时，会使缓存行无效。

注意：有两种情况下处理器不会使用缓存锁定。

1. 当操作的数据不能被缓存在处理器内部，或操作的数据跨多个缓存行时，则处理器会调用总线锁定。

2. 有些处理器不支持缓存锁定，对于 Intel 486 和 Pentium 处理器，就是锁定的内存区域在处理器的缓存行也会调用总线锁定。

CAS举例

CAS源码分析

JUC下的atomic类都是通过CAS来实现的，下面就以AtomicInteger为例来阐述CAS的实现。如下；下面我们看看AtomicIntger类的源码（此处“为了wzj”强烈建议学习的同学使用idea来看源码，以为ec看源码很不方便）

Unsafe是CAS的核心类，Java无法直接访问底层操作系统，而是通过本地（native）方法来访问。不过尽管如此，JVM还是开了一个后门：Unsafe，它提供了硬件级别的原子操作。

内部调用unsafe的getAndAddInt方法，在getAndAddInt方法中主要是看compareAndSwapInt方法：

 

上述为IDEA工具可见的相关方法，我们可以看到compareAndSwapInt方法前用了native修改，这个意思就是说他属于本地方法栈如果再发下深究，需要我们下载相关文件，OpenJDK，下载地址，也是从网上找的：https://www.cnblogs.com/dennyzhangdd/p/6734933.html

AtomicIntger.incrementAndGet()---unsafe--unsafe.cpp--atomic_windows_x86.inline.hpp--二进制--CUP  从java语言到C语言再到汇编指令最后是硬件CPU可以使用的二进制数据的转换过程。

有兴趣的同学可以按照以上步骤，深入了解一下。

CAS可以保证一次的读-改-写操作是原子操作，在单处理器上该操作容易实现，但是在多处理器上实现就有点儿复杂了。

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class CASDome1 {
    private static AtomicInteger atomicInteger = new AtomicInteger(1);

    public static void main(String args[]){
        Thread t1 = new Thread(()->{
            System.out.println(atomicInteger.compareAndSet(2,1));
            System.out.println("第一次修改："+atomicInteger.get());
        });
        t1.start();

        Thread t2 = new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(atomicInteger.compareAndSet(1,2));
            System.out.println("第二次修改："+atomicInteger.get());
        });
        t2.start();

        Thread t3 = new Thread(()->{
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(atomicInteger.compareAndSet(1,3));
            System.out.println("第三次修改："+atomicInteger.get());
        });
        t3.start();
    }

}

　　

缓存加锁：其实针对于上面那种情况我们只需要保证在同一时刻对某个内存地址的操作是原子性的即可。缓存加锁就是缓存在内存区域的数据如果在加锁期间，当它执行锁操作写回内存时，处理器不在输出LOCK#信号，而是修改内部的内存地址，利用缓存一致性协议来保证原子性。缓存一致性机制可以保证同一个内存区域的数据仅能被一个处理器修改，也就是说当CPU1修改缓存行中的i时使用缓存锁定，那么CPU2就不能同时缓存了i的缓存行

CAS缺点

CAS虽然高效地解决了原子操作，但是还是存在一些缺陷的，主要表现在三个方法：循环时间太长、只能保证一个共享变量原子操作、ABA问题。

循环时间太长

如果CAS一直不成功呢？这种情况绝对有可能发生，如果自旋CAS长时间地不成功，则会给CPU带来非常大的开销。在JUC中有些地方就限制了CAS自旋的次数，例如BlockingQueue的SynchronousQueue。

只能保证一个共享变量原子操作

看了CAS的实现就知道这只能针对一个共享变量，如果是多个共享变量就只能使用锁了，当然如果你有办法把多个变量整成一个变量，利用CAS也不错。例如读写锁中state的高地位

ABA问题

CAS需要检查操作值有没有发生改变，如果没有发生改变则更新。但是存在这样一种情况：如果一个值原来是A，变成了B，然后又变成了A，那么在CAS检查的时候会发现没有改变，但是实质上它已经发生了改变，这就是所谓的ABA问题。对于ABA问题其解决方案是加上版本号，即在每个变量都加上一个版本号，每次改变时加1，即A —> B —> A，变成1A —> 2B —> 3A。AtomicStampedReference类的版本控制

AQS

AbstractQueuedSychronizer（同步发射器），用来构建锁

与CAS相比，级别要高，但与Synchronized相比级别要低

基本思想

通过内置得到FIFO（先入先出，公平竞争原则）同步队列来完成线程争夺资源的管理。BlockingQueue。没记错的话zkr当前LIs7系统下的批处理就用到了这个类和线程池来实现的定时批处理

CLH同步队列

原理图：

自定义锁，读写分离锁

使用AQS定义一个锁，我们可以打开API看看AbstractQueuedSynchronizer的中文解释

CAS AQS子类定义为非公共内部帮助器，写锁的时候的一个帮助器，提供获取锁和释放锁的功能

acquire（int arg） 独占模式获取对象，忽略中断

acquereShared（int arg） 共享模式模式获取对象

tryAcquire（int arg） 试图获取独占模式获取对象

tryAcquireShared （int arg）试图获取共享模式模式获取对象

release（int arg） 释放独占模式获取对象

releaseShared（int arg） 释放共享模式获取对象

public class MyLock implements Lock {
    private Helper helper = new Helper();

    private class  Helper extends AbstractQueuedSynchronizer{
        //获取锁
        @Override
        protected boolean tryAcquire(int arg) {
            //获取状态
            int state = getState();//
            //使用CAS设置状态，使用CAS可以保证线程安全性，因为不是第一个节点，将不会占用独立线程
            if(compareAndSetState(0,arg)){
                //设置独立的主线程,将当前线程给当做主线程进行start();
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }else if(getExclusiveOwnerThread()== Thread.currentThread()){ //可以重复加锁
                setState(getState()+arg);
                return true;
            }
            return false;
        }
        //释放锁

        @Override
        protected boolean tryRelease(int arg) {
            //获取状态
            int state = getState()-arg;
            boolean flag = false;
            if(state == 0){
                //释放当前占用的线程
                setExclusiveOwnerThread(null);
                setState(state);
                return true;
            }
            setState(state);
            return false;
        }

        public Condition newObjectCondition(){
            return new ConditionObject();
        }
    }
    @Override
    public void lock() {
        //获取锁
        helper.acquire(1);
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        //获取中断的锁
        helper.acquireInterruptibly(1);
    }

    @Override
    public boolean tryLock() {
        //尝试获取锁
        return helper.tryAcquire(1);
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        //再time时间内，尝试获取锁，如果获取不到将抛出中断异常
       return  helper.tryAcquireNanos(1,unit.toNanos(time ));
    }

    @Override
    public void unlock() {
        //释放锁
        helper.release(1);
    }

    @Override
    public Condition newCondition() {
        //返回一个条件对象
       return helper.newObjectCondition();
    }
}

　　

public class MyLockTest {
    int m = 0;
    int next(){
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
           new RuntimeException("睡眠失败");
        }
        return ++m;
    }
    public static void main(String args[]){
        MyLockTest test = new MyLockTest();
        Thread[] thread = new Thread[30];
        for (int i = 0; i <30 ; i++) {
            thread[i] = new Thread(()->
               System.out.println(test.next())
            );
            thread[i].start();
        }
    }
}

出现线程不安全状态。

下面我们使用自己写的MyLock进行加锁

public class MyLockTest {
    private MyLock myLock = new MyLock();
    int m = 0;
    int next(){
        myLock.lock();
        try {
            return ++m;
        }  finally {
            myLock.unlock();
        }
    }

    public static void main(String args[]){
        MyLockTest test = new MyLockTest();
        Thread[] thread = new Thread[30];
        for (int i = 0; i <30 ; i++) {
            thread[i] = new Thread(()->
               System.out.println(test.next())
            );
            thread[i].start();
        }
    }
}

　　

自定义的锁一定要有个特性叫做可重入性，否则就会出现死锁，A方法包含B方法，A方式是存在锁的B方法也是存在锁的，这种情况如果是非可重入锁，将会死锁。这个特性体现在tryAcqure方法中如果state！=0会走 getExclusiveOwnerThread()== Thread.currentThread()

public void a(){
        myLock.lock();
        System.out.println("这是A");
        b();
        myLock.unlock();
    }

    public void b(){
        myLock.lock();
        System.out.println("这是B");
        myLock.unlock();
    }

　　

ReentrantLock与我们上面自定锁所功能一样，但是存在更牛逼的公平锁与非公平锁区别

再创建对象的时候如果参数是true则说明是公平锁，不传参数或者false则为非公平锁

ReentrantReaderWriter

读取者优先或者写入者优先，强加给锁的访问排序性，但是，它确实支持可选的公平策略

读写锁中存在一个重要的考点就是，锁降级，需要在writer.unlock之前添加reader.lock。个人理解就是，从读锁到写锁是正常顺序（先解锁再读锁，此时数据切换空隙是可见的），但是从写锁到读锁时候切换的时候没有任何加锁，此时其他线程会趁虚而入，获取当锁的state并修改。就像电视剧中经常出现的晚上值班换岗经常有贼在此时趁虚而入。详情可以看看老邱的博客，感觉写的比较直白：https://blog.csdn.net/qiuwenjie123/article/details/79950532

并发工具

CountdownLatch

public class CountDownLatchTest {
    private static List<String> list = Arrays.asList("东方航空","北方航空","海南航空");
    private static CountDownLatch cdl = new CountDownLatch(list.size());

    public static void main(String args[]){
        String init = "北京";
        String down = "悉尼";
        Thread [] threads = new Thread[list.size()];
        CountDownLatch cdl = new CountDownLatch(list.size());

        for (int i = 0; i < list.size(); i++) {
            String name = list.get(i);
            threads[i] = new Thread(()->{
                try {
                    int value = new Random().nextInt(10);
                    TimeUnit.SECONDS.sleep(value);
                    System.out.printf("%s从%s飞往%s的航班,存在的飞机票数量是%s"+"
",name,init,down,value);
                    cdl.countDown();//在此处会根据完成线程数量的设定计数
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            threads[i].start();
        }

        try {
            cdl.await();//计数达到了后直接唤醒主线程
            System.out.println("都返回给客户！");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}

　　

CyclicBarries

public class CyclicBarrierTest {
    public static void main(String args[]){
        CyclicBarrier cyclinBarrier = new CyclicBarrier(5);
        for (int i = 0; i < 5; i++) {

            new Thread(()->{
                try {
                    TimeUnit.SECONDS.sleep(new Random().nextInt(10));
                    System.out.println("第["+Thread.currentThread().getName()+"]号赛道的选手准备好了");
                        cyclinBarrier.await();

                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
                System.out.println("选手开始，跑！");
            }).start();

        }

    }
}

　　

semaphore：就像停车场一样，50个停车位，100辆车。

public class DiningRoomTest {
	//定义两个食堂窗口,公平队列，必须排队
	static Semaphore semaphore = new Semaphore(3,true);
	
	//定义学生
	 private static class Student implements Runnable{
		 private int type;
		 private Semaphore stuSemaphore;
		 private String name;
		 
		 private Student(){
			 
		 }
		 
		 private Student(int type,Semaphore stuSemaphore,String name){
			 this.type = type;
			 this.stuSemaphore = stuSemaphore;
			 this.name = name;
		 }

		 	
		public void run() {
			// TODO Auto-generated method stub
			switch (type) {
			case 0:   	//此种学生比较有耐心，一直在此等带食堂窗口打饭
				try {
					//这里面的人按照顺序一直的等待
					stuSemaphore.acquireUninterruptibly();
					int switeTime = new Random().nextInt(5);
					TimeUnit.SECONDS.sleep(switeTime);//打饭时间为0-5秒不等
					System.out.println("名字为："+this.name+"同学,打饭用时"+switeTime+"秒！");
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				stuSemaphore.release();//打完将释放
				break;

			case 1:		//此种学生比没有耐心，如果5分钟还打不上饭，就去商店买吃的
				try {
					if(stuSemaphore.tryAcquire(5, TimeUnit.SECONDS)){
						
							//这里面的同学只能等待8分钟，如果还没轮到他们，他就直接去商店买泡面吃
							int switeTime = new Random().nextInt(5);
							TimeUnit.SECONDS.sleep(switeTime);//打饭时间为0-5秒不等
							System.out.println("名字为："+this.name+"同学,打饭用时"+switeTime+"秒！");
							stuSemaphore.release();//打完将释放
						
					}else{
						System.out.println("名字为："+this.name+"同学,因不想等待直接去商店了");
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				break;
				
			case 2:		//此种学生比较有耐心，但是可能被老师叫走，商量班级里的事情
				
				try {
					//这里面的同学如果被老师叫走则就不去吃饭了
					stuSemaphore.acquire();
					int switeTime = new Random().nextInt(4);;
					TimeUnit.SECONDS.sleep(switeTime);//打饭时间为0-5秒不等
					stuSemaphore.release();//打完将释放
					System.out.println("名字为："+this.name+"同学,打饭用时"+switeTime+"秒！");
				} catch (InterruptedException e) {
					System.out.println("名字为："+this.name+"同学,被老师叫走啦！");
				}
				
				break;
			}
		}
	}
	 
	 
	 public static void main(String[] args) {
		 for (int i = 1; i <= 10; i++) {
			 Thread thread = new Thread(new Student(0,DiningRoomTest.semaphore,"同学"+i));
			 thread.start();
		}
		 
		 for (int i = 11; i <= 15; i++) {
			 Thread thread = new Thread(new Student(1,DiningRoomTest.semaphore,"同学"+i));
			 thread.start();
		}
		 
		 for (int i = 16; i <= 20; i++) {
			 Thread thread = new Thread(new Student(2,DiningRoomTest.semaphore,"同学"+i));
			 thread.start();
			 try {
				TimeUnit.SECONDS.sleep(2);
			} catch (InterruptedException e) {
			}
			 thread.interrupt();
		}
	}

}