2022-02-17 17:52 华北科技学院 Java

关注

JAVA 锁

多并发之锁

创建时间2022/02/11 11:42:14

1.synchronized

1.1锁的粒度

同步代码块，锁的是synchronized括号里的对象
同步方法，锁的是当前实例对象this
静态同步方法，锁的是当前类的Class对象

1.2 synchronized的实现机制

synchronized的锁存储在对象头里。对象头的mark work中存储了锁的信息，包括锁的标志位和状态

java的对象头结构：

长度	存储内容	说明
32/64bit	Mark work	存储hashcode或者锁的信息
32/64bit	Class Metadata address	引用对象的地址
32/64bit	Array length	数组长度（只有数组有）

Mark work默认结构

锁状态	25bit	4bit	1bit（是否偏向锁）	2bit锁标志位
无锁状态	hashcode	对象分代年龄	0	01

Mark word演化结构

alt

1.3 锁升级

锁分为无锁状态，偏向锁，轻量级，重量级锁，这几个状态会随着竞争情况逐渐升级。锁。锁只能升级，不能降级，锁升级机制是为了提高了获得锁和释放锁的效率

1.4 偏向锁

MarkWord中存储内容为：线程ID—epoch—1—01

不存在竞争时，当一个线程访问同步代码块并获取锁时，会在对象头的markword中和栈帧中记录当前线程ID。当下次访问时，会检查markwork中是否有当前线程id，如果有直接获取锁，不需要进行CAS操作。 alt 访问同步代码块，检查markwork中是否含有当前线程id，没有就CAS替换Mark work，将Markword中的线程ID指向自己并吧是否偏向锁位置为1，然后执行同步体。当第二个线程来访问时，检查markwork中的没有线程2的ID后进行CAS替换Markwork操作，失败后会暂停线程1然后解锁将线程ID置空，撤销偏向锁。

1.5 轻量级锁

Mark work存储内容为指向栈帧记录的指针—00

当线程访问同步代码时，线程会创建锁记录空间，并把对象的Markwork复制到锁记录空间中，官方称为Displaced Mark Work。然后线程尝试CAS将对象头中的Markwork替换为指向栈帧记录的指针。如果成功获得锁，如果失败，表示其他线程正在竞争然后采取CAS+自旋的操作。

释放锁时，CAS操作将Displaced Mark Work替换回到对象头中，如果成功则表示没有竞争发生，如果失败则表示竞争激烈，锁膨胀为重量级锁

alt

2. AQS

独享方式

独享锁获取锁的过程

首先直接调用tryAcquire（）抢占锁，不行以后调用addWairter（）加入等待队列，并进入阻塞抢锁

具体抢锁机制：如果为等待队列第二名，就直接抢锁，否则判断是否需要进入阻塞（前节点为SIGNAL都需要阻塞）。

如果抢成功了，将自己设为头结点，与前节点断开连接，呼叫GC

//独享锁获取锁
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}
//添加队列
private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}
//抢锁
final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

//独享锁释放锁
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

共享方式

//acquireShared() 尝试一次抢锁，如果失败进去doAcquireShared，一般失败的原因是因为前面有独占锁
public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}
//持续抢占锁的方式
private void doAcquireShared(int arg) {
final Node node = addWaiter(Node.SHARED);//加入等待队列
boolean failed = true;
try {
    boolean interrupted = false;
    for (;;) {
        final Node p = node.predecessor();
        if (p == head) {
            //如果为第二个节点就开始抢
            int r = tryAcquireShared(arg);
            if (r >= 0) {
            //如果抢成功以后，设置为头结点，并继续向后延续
                setHeadAndPropagate(node, r);
                p.next = null; // help GC
                if (interrupted)
                    selfInterrupt();
                failed = false;
                return;
            }
        }
        if (shouldParkAfterFailedAcquire(p, node) &&
            parkAndCheckInterrupt())
            interrupted = true;
    }
} finally {
    if (failed)
        cancelAcquire(node);
}
}
//setHeadAndPropagate()
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
setHead(node);//设置为头结点

if (propagate > 0 || h == null || h.waitStatus < 0 ||
    (h = head) == null || h.waitStatus < 0) {
    Node s = node.next;
    if (s == null || s.isShared())//如果下一个节点为共享占有
        doReleaseShared();//唤醒下一个节点或者设置传播状态
}
}
private void doReleaseShared() {

for (;;) {
    Node h = head;
    if (h != null && h != tail) {
        int ws = h.waitStatus;
        if (ws == Node.SIGNAL) {
        //如果需要唤醒，就唤醒。并将状态设置为0
            if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                continue;            // loop to recheck cases
            unparkSuccessor(h);
        }
        //设置状态为 PROPAGATE 状态，保证能够持续延续
        else if (ws == 0 &&
                 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
            continue;                // loop on failed CAS
    }
    if (h == head)                   // loop if head changed
        break;
}
}

3. LOCK

3.1 Lock接口

    //加锁
    void lock();
    //解锁
    void unlock();
    //加锁（可中断） 当获取锁的过程中被中断，会抛出中断异常，并释放锁
    void lockInterruptibly() throws InterruptedException;
    //加锁（非阻塞） 尝试获取锁，若此时锁没有被其他锁占用，则会成功获取锁
    boolean tryLock();
    //加锁（超时） 在指定的截止时间之前获取锁
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    //获取等待通知组件
    Condition newCondition();

3.2 ReentrantLock实现

//抽象类Sync
abstract static class Sync extends AbstractQueuedSynchronizer
//公平锁
static final class FairSync extends Sync
//非公平锁
static final class NonfairSync extends Sync

Sync的实现

//非公平抢锁方法
/*
创建当前线程,获取当前锁的冲入次数
如果次数为0,则代表第一次进入,CAS更换锁state的次数,将当前锁的专属线程更改为当前线程
如果不为0,则累加
*/
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}
//释放锁
/*
用state减掉释放层数
如果为0,则将专属线程置空
然后更改state
*/
protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

FairSync的实现

//加锁
final void lock() {
    acquire(1);
}

//实现tryAcquire() 
**//和NoFairSync相似,唯一区别在于在cas更换state前检查了一下前面有没有元素****
**  protected final boolean tryAcquire(int acquires) {
      final Thread current = Thread.currentThread();
      int c = getState();
      if (c == 0) {
          if (!hasQueuedPredecessors() &&
              compareAndSetState(0, acquires)) {
              setExclusiveOwnerThread(current);
              return true;
          }
      }
      else if (current == getExclusiveOwnerThread()) {
          int nextc = c + acquires;
          if (nextc < 0)
              throw new Error("Maximum lock count exceeded");
          setState(nextc);
          return true;
      }
      return false;
  }
}

NoFairSync的实现

//加锁
**//与FairSync相比在进行acquire之前先进行了一次抢锁，也是不公平的体现
**final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

//tryAcquire 的重写
protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}

4. ReadWriteLock

4.1 ReadWriteLock 接口

//写锁
Lock readLock();
//读锁
Lock writeLock();

4.2 ReentrantReadWriteLock

//公平锁
static final class FairSync extends Sync{}
//非公平锁
static final class NonfairSync extends Sync{}
//锁
abstract static class Sync extends AbstractQueuedSynchronizer{}
//读锁（共享锁）
public static class ReadLock implements Lock, java.io.Serializable{}
//写锁（独占锁）
public static class WriteLock implements Lock, java.io.Serializable

FairSync

    static final class FairSync extends Sync {
        private static final long serialVersionUID = -2274990926593161451L;
        final boolean writerShouldBlock() {
            return hasQueuedPredecessors();//公平锁，按顺序来前面有节点就阻塞
        }
        final boolean readerShouldBlock() {
            return hasQueuedPredecessors();//公平锁，按顺序来前面有节点就阻塞
        }
    }

NoFairSync

    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -8159625535654395037L;
        final boolean writerShouldBlock() {
            return false; // writers can always barge（写锁总是不阻塞）
        }
        final boolean readerShouldBlock() {
            return apparentlyFirstQueuedIsExclusive();//检查第一个排队的节点是否为写锁
        }
    }

Sync

          **//tryAcquire（）**
          protected final boolean tryAcquire(int acquires) {
            /*
             * Walkthrough:
             * 1. If read count nonzero or write count nonzero
             *    and owner is a different thread, fail.
             * 2. If count would saturate, fail. (This can only
             *    happen if count is already nonzero.)
             * 3. Otherwise, this thread is eligible for lock if
             *    it is either a reentrant acquire or
             *    queue policy allows it. If so, update state
             *    and set owner.
             */
            Thread current = Thread.currentThread();//获取当前线程
            int c = getState();//获取加锁次数
            int w = exclusiveCount(c);//获取独占次数
            if (c != 0) {
                // (Note: if c != 0 and w == 0 then shared count != 0)
                if (w == 0 || current != getExclusiveOwnerThread())//如果是读锁或者当前线程不是自己的则失败
                    return false;
                if (w + exclusiveCount(acquires) > MAX_COUNT)//若果加锁后超过最大加锁次数则失败
                    throw new Error("Maximum lock count exceeded");
                // Reentrant acquire
                setState(c + acquires);//加锁成功
                return true;
            }
            //如果是第一次加锁，判断是否应该阻塞
            //不公平锁永远不阻塞，公平锁判断前面有没有节点
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))//如果加锁失败则返回false
                return false;
             //第一次加锁更改专属线程为当前线程
            setExclusiveOwnerThread(current);
            return true;
        }
        //tryAcquireShared
        protected final int tryAcquireShared(int unused) {
            /*
             * Walkthrough:
             * 1. If write lock held by another thread, fail.
             * 2. Otherwise, this thread is eligible for
             *    lock wrt state, so ask if it should block
             *    because of queue policy. If not, try
             *    to grant by CASing state and updating count.
             *    Note that step does not check for reentrant
             *    acquires, which is postponed to full version
             *    to avoid having to check hold count in
             *    the more typical non-reentrant case.
             * 3. If step 2 fails either because thread
             *    apparently not eligible or CAS fails or count
             *    saturated, chain to version with full retry loop.
             */
            Thread current = Thread.currentThread();//获取当前线程
            int c = getState();//获取当前锁的状态
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)//当前有写锁但不是自己，则失败
                return -1;
            int r = sharedCount(c);
            //判断读锁应不应该阻塞
            //公平锁检查前面有没有节点，不公平锁检查第一个排队的锁是不是写锁
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {

                if (r == 0) {
                    firstReader = current;
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                    firstReaderHoldCount++;
                } else {
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current))
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                }
                return 1;
            }
            return fullTryAcquireShared(current);
        }
**        //tryRelease（）
**         protected final boolean tryRelease(int releases) {
            if (!isHeldExclusively())//如果不是独占锁抛出异常
                throw new IllegalMonitorStateException();
            int nextc = getState() - releases;//获取减完以后的锁状态
            boolean free = exclusiveCount(nextc) == 0;
            //如果没有锁了，则吧专属线程置空
            if (free)
                setExclusiveOwnerThread(null);
            //更改锁状态
            setState(nextc);
            return free;
        }
        **//trySharedRelease（）**
        protected final boolean tryReleaseShared(int unused) {
            Thread current = Thread.currentThread();
            for (;;) {
                int c = getState();
                int nextc = c - SHARED_UNIT;
                if (compareAndSetState(c, nextc))
                    // Releasing the read lock has no effect on readers,
                    // but it may allow waiting writers to proceed if
                    // both read and write locks are now free.
                    return nextc == 0;
            }
        }

ReentrantReadWriteLock加锁过程

alt