AQS之同步器

Lock lock = new ReentrantLock();
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        lock.lock();
        System.out.println(Thread.currentThread().getName() + " " + LocalDateTime.now());
        sleep(TimeUnit.SECONDS, 2);
        lock.unlock();
    }).start();
}

---------------------------------执行结果------------------------------------
Thread-0 2020-01-01T17:12:24.678
Thread-1 2020-01-01T17:12:26.681
Thread-2 2020-01-01T17:12:28.681

CountDownLatch countDownLatch = new CountDownLatch(3);
new Thread(() -> {
    sleep(TimeUnit.MILLISECONDS, 80);
    System.out.println(Thread.currentThread().getName() + " Finished");
    countDownLatch.countDown();
}).start();
new Thread(() -> {
    sleep(TimeUnit.MILLISECONDS, 50);
    System.out.println(Thread.currentThread().getName() + " Finished");
    countDownLatch.countDown();
}).start();
new Thread(() -> {
    sleep(TimeUnit.MILLISECONDS, 60);
    System.out.println(Thread.currentThread().getName() + " Finished");
    countDownLatch.countDown();
}).start();
new Thread(() -> {
    sleep(TimeUnit.MILLISECONDS, 60);
    System.out.println(Thread.currentThread().getName() + " Finished");
    countDownLatch.countDown();
}).start();
countDownLatch.await();
System.out.println("All Finished");


---------------------------------执行结果------------------------------------
Thread-1 Finished
Thread-2 Finished
Thread-3 Finished
All Finished
Thread-0 Finished

CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> System.out.println("阻塞解除之后执行一些逻辑 " + LocalDateTime.now()));
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        try {
            System.out.println(Thread.currentThread().getName() + " doXX");
            cyclicBarrier.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();

    new Thread(() -> {
        try {
            System.out.println(Thread.currentThread().getName() + " doXX");
            cyclicBarrier.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }).start();

    sleep(TimeUnit.SECONDS, 2);
}

---------------------------------执行结果------------------------------------
Thread-0 doXX
Thread-1 doXX
阻塞解除之后执行一些逻辑 2020-01-01T14:19:15.735
Thread-2 doXX
Thread-3 doXX
阻塞解除之后执行一些逻辑 2020-01-01T14:19:17.706
Thread-4 doXX
Thread-5 doXX
阻塞解除之后执行一些逻辑 2020-01-01T14:19:19.706

Semaphore semaphore = new Semaphore(2);
for (int i = 0; i < 4; i++) {
    new Thread(() -> {
        try {
            semaphore.acquire();
            System.out.println(Thread.currentThread().getName() + "访问资源");
            sleep(TimeUnit.MILLISECONDS, 10 + new Random().nextInt(10));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println(Thread.currentThread().getName() + "释放资源");
            semaphore.release();
        }

    }).start();
}

---------------------------------执行结果------------------------------------
Thread-1访问资源
Thread-0访问资源
Thread-0释放资源
Thread-2访问资源
Thread-1释放资源
Thread-3访问资源
Thread-2释放资源
Thread-3释放资源

StringBuilder data = new StringBuilder("data");
ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();

new Thread((() -> {
    // sleep一会,让写锁先持有锁
    sleep(TimeUnit.MILLISECONDS, 10);

    readWriteLock.readLock().lock();
    System.out.println(Thread.currentThread().getName() + "获得读锁 " + LocalDateTime.now());
    System.out.println(Thread.currentThread().getName() + data);
    readWriteLock.readLock().unlock();
}), "readLock ").start();

new Thread((() -> {
    readWriteLock.writeLock().lock();
    System.out.println(Thread.currentThread().getName() + "获得写锁 " + LocalDateTime.now());
    sleep(TimeUnit.SECONDS, 3);
    data.append("666");

    // 锁降级 这里这样子使用降级感觉没有什么意思,那锁降级一般用到什么场景
    readWriteLock.readLock().lock();
    System.out.println(Thread.currentThread().getName() + data);

    readWriteLock.writeLock().unlock();
}), "writeLock ").start();

public ReentrantReadWriteLock() {
    this(false);
}

// 可以在创建ReentrantReadWriteLock时选择公平模式还是非公平模式
public ReentrantReadWriteLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
    readerLock = new ReadLock(this);
    writerLock = new WriteLock(this);
}

static final int SHARED_SHIFT   = 16;
static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

/** Returns the number of shared holds represented in count  */
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }

/** Returns the number of exclusive holds represented in count  */
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

public void lock() {
    sync.acquireShared(1);
}

public final void acquireShared(int arg) {
    if (tryAcquireShared(arg) < 0)
        doAcquireShared(arg);
}

protected final int tryAcquireShared(int unused) {
    Thread current = Thread.currentThread();
    int c = getState();

    // 其实这里就涉及到锁降级,如果当前已经有写锁,返回-1,将它加入到阻塞队列, 否则继续尝试获取读锁
    if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
        return -1;

    int r = sharedCount(c);
    // readerShouldBlock主要是针对公平锁和非公平锁, `c + SHARED_UNIT`是因为共享锁用的是高16位
    if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) {
        if (r == 0) {
            // firstReader是把读锁状态从0变成1的那个线程
            firstReader = current;
            firstReaderHoldCount = 1;
        } else if (firstReader == current) {
            // 锁重入
            firstReaderHoldCount++;
        } else {
            // cachedHoldCounter是上一个获取锁成功的线程
            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;
    }

    // 上面的if失败了,即通过自旋获取锁
    return fullTryAcquireShared(current);
}

final int fullTryAcquireShared(Thread current) {
    HoldCounter rh = null;
    for (;;) {
        int c = getState();
        if (exclusiveCount(c) != 0) {
            if (getExclusiveOwnerThread() != current)
                return -1;
            // else we hold the exclusive lock; blocking here
            // would cause deadlock.
        } else if (readerShouldBlock()) {
            // Make sure we're not acquiring read lock reentrantly
            if (firstReader == current) {
                // assert firstReaderHoldCount > 0;
            } else {
                if (rh == null) {
                    rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current)) {
                        rh = readHolds.get();
                        if (rh.count == 0)
                            readHolds.remove();
                    }
                }
                if (rh.count == 0)
                    return -1;
            }
        }
        if (sharedCount(c) == MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
        if (compareAndSetState(c, c + SHARED_UNIT)) {
            if (sharedCount(c) == 0) {
                firstReader = current;
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;
            } else {
                if (rh == null)
                    rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                rh.count++;
                cachedHoldCounter = rh; // cache for release
            }
            return 1;
        }
    }
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

protected final boolean tryReleaseShared(int unused) {
    Thread current = Thread.currentThread();

    // 如果 firstReader 是当前线程
    if (firstReader == current) {
        // assert firstReaderHoldCount > 0;

        // 并且firstReaderHoldCount == 1,说明释放锁之后需要重置firstReader
        if (firstReaderHoldCount == 1)
            firstReader = null;
        else
            firstReaderHoldCount--;
    } else {
        HoldCounter rh = cachedHoldCounter;
        if (rh == null || rh.tid != getThreadId(current))
            rh = readHolds.get();
        int count = rh.count;
        if (count <= 1) {
            readHolds.remove();
            if (count <= 0)
                throw unmatchedUnlockException();
        }
        --rh.count;
    }
    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;
    }
}

protected final boolean tryAcquire(int acquires) {
    Thread current = Thread.currentThread();
    int c = getState();
    int w = exclusiveCount(c);

    // state != 0 && w==0 说明有读锁
    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;
    }

    // writerShouldBlock用于确定公平和非公平模式
    if (writerShouldBlock() ||
        !compareAndSetState(c, c + acquires))
        return false;
    setExclusiveOwnerThread(current);
    return true;
}

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;
}