Java多线程并发——CAS和AQS

r囧r小猫 2022-10-10 01:46 149阅读 0赞

多核CPU、多线程的场景下，一起学习Java如何保证程序原子性，有序性，以及数据完整性等特性。

## CAS ##

**Compare And Swap**

原子操作，更新之前，比较期望值，如果是期望值的话，写数据，否则不写数据，更新失败。

Java的CAS操作调用的是unsafe本地Native方法，通过使用CPU相关指令来达到原子性操作，包括多核CPU下的原子操作。

通常为保证更新成功，操作需要自旋。即不断的尝试CAS更新，直到更新成功。如AtomicInteger中的一段代码：

public final boolean compareAndSet(int expect, int update) { 
            return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
        }

自旋更新

public final int getAndUpdate(IntUnaryOperator updateFunction) { 
            int prev, next;
            do { 
                prev = get();
                next = updateFunction.applyAsInt(prev);
            } while (!compareAndSet(prev, next));
            return prev;
        }

**ABA问题**，CompareAndSwap的值从A变为B，再由B变为A，这种情况下，CAS认为值没有变，但其实是变了的，需要使用版本号来解决此问题，Atomic使用AtomicStampedReference来解决。

**特点**  
CAS认为锁竞争不激烈，更新成功概率高，在激烈竞争的情况下，更新成功概率降低，自旋时间变长，影响服务器性能

只能保证一个共享变量的原子性操作，多个共享变量时，需要将多个共享变量合成一个共享变量AtomicReference

## AQS ##

**AbstractQueuedSychronizer**

抽象队列同步，实现锁和同步机制的基础框架，RetreentLock RetreentLockReadWriteLock Semaphore CountDownLatch等实现基于AQS

AQS维护一个volatile修饰的state字段，来控制多个线程之间的独占或者共享同步状态。state值的修改通过CAS的方式。  
AQS维护了一个FIFO的队列，线程获取同步状态失败时，构建一个Node并添加到队列尾部tail（通过CAS实现），并阻塞当前线程，当其他释放同步状态时，判断队列的head是否为空，不为空head节点获得锁

通过自旋的方式获得锁或者队列排队的方式获取锁。

## sychronized ##

sychronized为隐式锁，编也就是译器自动加锁，解锁，并在异常情况下自动解锁。

sychronized使用对象锁，类锁实现，对象锁是对某一象加锁，类锁是对某一个类加锁。

类和对象的wait和notifiy方法，wait方法释放锁并等待锁，锁notifiy之后，有可能重新获得锁。

*  对静态方法加锁，等于对类加锁，影响程序性能，其他对类加锁的线程都需要等待。
 *  对实例方法加锁，等于对对象加锁，影响程序性能，其他对对象加锁的线程都需要等待

使用sychronized，一般做法是定义功能单一的类或对象，使用其类锁或对象锁，对代码块临界区加锁。

临界区是指需要同步控制的代码块。

**sychronized缺点**  
1，不可响应中断，线程请求锁时，没有打断机制，可能造成死锁  
2，请求锁没有超时机制，一直阻塞等待  
3，不能尝试获取锁，只能阻塞等待。（尝试获取锁，tryLock，没获取到直接返回，继续执行其他逻辑）

## Lock ##

接口，锁的顶级接口  
lock vs sychronized

**优点**  
1，提供了tryLock，请求锁不会一直等待，引入请求锁的超时机制  
2，更加灵活，可以根据代码的不同条件来决定释放锁或者请求锁。  
3，lockInterruptibly，支持中断阻塞线程，避免死锁发生

**缺点**  
显式的锁，需要手动的获得锁，关锁，处理异常情况下的锁问题

## ReentrantLock ##

Lock的实现类，基于AQS实现

1.  **支持公平获取锁和非公平获取锁**

**公平锁**加锁过程：当state=0且队列中没有等待时，尝试CAS获取锁，没有获取到锁，添加到队列中。当队列中有等待时，也加入队列中等待，直到线程变为队列Head的时候自旋获取锁。

**非公平锁**加锁过程：直接尝试CAS获取锁，没有获取到锁，加入到同步队列中，一次获取锁，这时的非公平体现在head和新的线程不公平竞争，但是在同步队列中的还是要依次获得锁。ReentrantLock默认是非公平锁。

private ReentrantLock lock = new ReentrantLock(); // non fair，非公平锁
    
    private ReentrantLock lock = new ReentrantLock(true); // fair，公平锁
    
    private ReentrantLock lock = new ReentrantLock(false); // no fair，非公平锁

1.  **尝试一次性非阻塞获取锁**，提高编码灵活性

lock.tryLock();

1.  **支持超时机制**，超过时间没有获取锁，直接返回。避免死锁。

lock.tryLock(10, TimeUnit.SECONDS);

1.  **支持可中断获取锁**，和lock()方法获取锁的方式相比，区别在于，lockInterruptibly()获取锁的过程可以被打断，其他线程调用了该线程的interrupt方法后，该线程不再尝试获取锁，而是执行线程中断，抛出InterruptedException，而lock()比较头铁，还会一直尝试获取锁，获取锁后才执行线程中断逻辑。

lock.lockInterruptibly();

1.  **可以监听不同condition等待条件**

condition，类似于Object和对象的，notify和wait方法，有等待和通知的功能。  
如：一只单纯的牛，每天就eat，sleep，work三件事情，分别由三个线程控制，每个线程分别不断的尝试eat，sleep，work，且在每个活动中互不影响，但是牛的主人规定牛在eat之后，才能sleep，在sleep后才能work，在work之后才能eat。

public class PureCow { 
        private ReentrantLock lock = new ReentrantLock();
    
        private Condition eatCondition = lock.newCondition();
        private Condition sleepCondition = lock.newCondition();
        private Condition workCondition = lock.newCondition();
    
        String status = "eat";
    
        public void eat(){ 
            lock.lock();
            System.out.println(Thread.currentThread().getName() + " thread get lock,start eat");
            try{ 
                if(!status.equals("eat")) { 
                    eatCondition.await();
                }
                System.out.println("cow eatting......");
                // 模拟 eat
                Thread.sleep(3000);
                status = "sleep";
                sleepCondition.signal();
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                lock.unlock();
            }
        }
    
        public void sleep(){ 
            lock.lock();
            System.out.println(Thread.currentThread().getName() + " thread get lock,start sleep");
            try{ 
                if(!status.equals("sleep")) { 
                    sleepCondition.await();
                }
                // 模拟 sleep
                System.out.println("cow sleepping......");
                Thread.sleep(3000);
                status = "work";
                workCondition.signal();
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                lock.unlock();
            }
        }
    
        public void work(){ 
            lock.lock();
            System.out.println(Thread.currentThread().getName() + " thread get lock,start work");
            try{ 
                if(!status.equals("work")) { 
                    workCondition.await();
                }
                // 模拟 work
                System.out.println("cow working......");
                Thread.sleep(3000);
                status = "eat";
                eatCondition.signal();
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                lock.unlock();
            }
        }
    
        public static void main(String[] args) { 
            PureCow cow = new PureCow();
            Thread t1 = new Thread( () -> { 
                while(true){ 
                    cow.eat();
                }
            }, "eat");
    
            Thread t2 = new Thread( () -> { 
                while(true){ 
                    cow.sleep();
                }
            }, "sleep");
            Thread t3 = new Thread( () -> { 
                while(true) { 
                    cow.work();
                }
            }, "work");
            t1.start();
            t2.start();
            t3.start();
        }
    }

## ReentrantReadWriteLock ##

**ReadWriteLock**，维护一对关联的锁，一个是针对只读操作的读锁，一个针对写操作写锁。

读锁可以线程读操作共享，写锁线程独占。

在操作共享数据时，跟独占锁相比较，ReadWriteLock支持更高的并发。

ReentrantReadWriteLock是ReentrantLock和ReadWriteLock的结合实现

**特点**

*  没有强加读写优先权
 *  支持公平和非公平策略，默认非公平模式
 *  重入，读锁可以重入读锁，写锁可以重入写锁，读锁可以重入写锁，但是写锁不能重入读锁，
 *  锁可以降级，写锁可以降级为读锁，因为读锁共享，有共享变为独有，代价比较大

读写锁的state状态，使用int类型的state的高16位标识读锁，低16位标识写锁

**读锁加锁过程**

1.  获取当前线程，判断写锁state是否为0，不为0说明有写锁，判断持有锁的是不是当前线程，如果不是当前线程，返回-1，加锁失败。
2.  判断读锁，已加锁次数小于65536，且不需要等待，尝试加锁。公平情况下队列中有等待读锁时需要等待，且非公平竞争情况下，队列中有等待的写锁需要等待。
3.  加锁成功后，修改锁被持有的数量，如果是第一个持有线程，修改第一个持有锁的线程，第一个持有锁的线程的持有次数。如果不是第一个持有的线程，在线程ThreadLocal中记录持有锁的次数，返回成功
4.  如果尝试加锁失败，自旋加锁。

**写锁加锁过程**

1.  获取当前线程，如果state等于0，没有锁，尝试CAS加锁，加锁成功后设置锁的独占线程。
2.  如果state不等于0，获取独占锁数量，如果等于0，说明有读锁，判断是不是当前线程，不是当前线程加锁失败，否则尝试CAS加锁，成功后设置锁的独占线程

锁降级和锁重入，读锁可以重入写锁（写锁中重新获取读锁），独占锁到共享锁比较容易，而共享锁转为独占锁比较难。 所以写锁不可以重入读锁，真实运行写锁重入读锁不会异常，而是一直卡住获取不到锁。

/** * 读锁重入写锁，写锁中重新获取读锁 */
        public void readReentryWrite(){ 
            writeLock.lock();
            try{ 
                System.out.println(Thread.currentThread().getName()+" get readLock");
                Thread.sleep(3000);
                System.out.println(Thread.currentThread().getName()+" do write");
                readLock.lock();
    
                System.out.println(Thread.currentThread().getName()+" do read");
                System.out.println(Thread.currentThread().getName()+" release readLock");
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                readLock.unlock();
                writeLock.unlock();
            }
        }
        
        /** * 锁降级，程序的后半段，writeLock.unlock()之后，线程的锁变为读锁 */
        public void degrade(){ 
            writeLock.lock();
            try{ 
                System.out.println(Thread.currentThread().getName()+" get readLock");
                Thread.sleep(3000);
                System.out.println(Thread.currentThread().getName()+" do write");
                readLock.lock();
                writeLock.unlock();
                System.out.println(Thread.currentThread().getName()+" do read");
                System.out.println(Thread.currentThread().getName()+" release readLock");
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                readLock.unlock();
                if (writeLock.isHeldByCurrentThread()) { 
                    writeLock.unlock();
                }
            }
        }

## StampedLock ##

StampedLock  
优化了ReentrantReadWriteLock，优化了RRWL的饥饿问题，如读操作很多，写操作很少时，写操作饥饿问题。

具有三种不同模式来控制读写访问，StampedLock不可重入，不支持Condition等待，支持读锁写锁转换。

StampedLock的state由版本和模式组成，获取锁的时候，返回一个标识并控制锁状态的stamp，try操作会反馈0来表示无法获取锁，锁的转换和释放需要使用stamp作为参数，如果stamp与锁的状态不匹配，则失败。

三种模式是：  
写：写锁独占访问，可能阻塞等待，获取锁后返回一个stamp，用来释放锁，支持超时try的方式获取锁，如果某线程获得了写锁，其他线程就不会获取读锁，乐观读也会失败。

读：获取读锁锁，类似于ReentrantReadWriteLock，获取锁后返回一个stamp，用来释放锁。支持超时try的方式获取锁，读锁可以转化为写锁

乐观读：获取乐观读锁时，当没有写锁时返回非0值，获取锁成功。乐观读锁可以被写锁直接占用，使用乐观锁时需要校验stamp，如果已经被写锁占用，就需要转为读锁，再读取重新读数据。

## Semaphore ##

控制共享资源的访问，同时只有有限个（根据信号量的定义）访问资源

支持公平竞争和非公平竞争

锁是特殊的信号量，同时只有一个线程访问资源。

通过自旋的方式CAS获取锁，公平模式下队列中有等待返回。

## ThreadLocal ##

**特点**：

*  多线程下，以空间换时间，数据在线程的工作空间以副本的形式存在，线程之间不共享，没有多线程安全问题
 *  提供线程执行任意阶段访问对象或数据的方式

缺点：

*  解决不了多线程数据共享问题

**应用**  
服务中在ThreadLocal中存储Context上下文信息，HTTP请求和响应信息  
事务中，ThreadLocal中存储数据库连接，控制事务的提交回滚

**细节**  
强引用，正常直接引用，有强引用时，垃圾回收机制不会回收对象。  
弱引用，弱引用对象会被回收，不管空间是否足够。  
软引用，如果内存空间足够，不会回收软引用，否则回收软引用，软引用多用作内存敏感的高速缓存  
虚引用，随时会被回收，和RefrenceQueue联合使用，跟踪垃圾回收器的回收活动

Java的Thread类中有一个ThreadLocalMap对象，其中存储着key和Value，每个ThreadLocal对应Map中的一个key，value，ThreadLocal本身是key，需要存储的对象就是value，如果需要存储多个对象，多个ThreadLocal，ThreadLocal对key是弱引用，ThreadLocal remove后就会被垃圾回收机制回收。

**扩展**  
对比netty中的FastThreadLocal，使用FastThreadLocal时，对应的线程中存储FastThreadLocal对象的容器变为数组

需要特殊的线程实现，需要配合FastThreadLocalThread线程来使用，使用普通的Thread线程时，会变的更慢。

因为FastThreadLocal为了和Thread兼容，还增加了SlowThreadLocalMap实现。

**问题**  
内存泄漏，方法执行完成后，栈没有引用后，ThreadLocal被回收，但是ThreadLocalMap被线程引用，不会被回收，会导致内存泄漏  
ThreadLocal被static修饰后，延长了ThreadLocal的声明周期，也有可能造成不会被回收。

（完 ^\_^）