Article 18 AQS Shared locks, Semaphore, CountDownLatch, I heard database connection pool can be used!

Blog: bugstack.cn Github: github.com/fuzhengwei/…

Precipitation, share, grow, let yourself and others can gain something! 😄

One, foreword

How can learning Java grow by leaps and bounds?

Is not you have seen the rapid progress of others, but their own is very difficult!

In fact, there is no one day of rapid progress, nor a mouthful out of fat. Have more time accumulate over a long period of time, ceaselessly precipitate, finally can erupt, break bureau!

To take a simple example, what if you had written 400,000 lines of code by the time you graduated from college and still couldn’t find a job? But 400,000 lines per day is not a lot, and what matters is consistency.

2. Interview questions

Thanks for the plane, note! The east wind blows, the drums beat, no overtime, who is afraid of who! Hahaha, go find my big brother.

Thank plane: feed, eldest brother. My girlfriend is stuck in the interview. It’s hard for a strong man to lock, and I can’t lock!

Interviewer: You shouldn’t be able to ah, ask you a, based on AQS implementation of the lock are what?

Xie Airplane: Well, there’s ReentrantLock…

Interviewer: Anything else?

Xie Airplane: I can’t seem to remember, sync is not either!

Interviewer: Well, learning without thinking, summarizing, or taking notes. You can’t talk in an interview like this, but at the very least, you have to be in depth.

Thank plane: hey hey, remember. Come to my home to eat hot pot, details.

Shared locks and AQS

1. What are the locks based on AQS?

AQS (AbstractQueuedSynchronizer), is very important in Java and contract awarding a class, most of the implementation of the lock is implemented based on AQS, including:

ReentrantLock, reentrant lock. This is the lock we started with, and it’s the most commonly used lock. Is often compared to synchronized.
ReentrantReadWriteLock, read/write lock. Read locks are shared locks and write locks are exclusive locks.
Semaphore, semaphore lock. Mainly used to control the flow, such as: database connection pool to you assigned 10 links, so let you come to a link, connect to 10 has not been released, then you wait.
CountDownLatch, atresia. Four people hold a rafting boat and when it’s full, they Latch onto it.

In this chapter we mainly introduce Semaphore, Semaphore lock implementation, in fact, is introduced a shared lock on the use and source analysis.

2. Use Semaphore shared lock

Semaphore semaphore = new Semaphore(2.false); Permitting: semapore, fair: fair/unfair
for (int i = 0; i < 8; i++) {
    new Thread(() -> {
        try {
            semaphore.acquire();
            System.out.println(Thread.currentThread().getName() + "Crouch hole");
            Thread.sleep(1000L);
        } catch (InterruptedException ignore) {
        } finally{ semaphore.release(); }},"Squat Number:" + i).start();
}
Copy the code

Here we simulated a scene of toilet queuing and squatting in a high-speed service area. Due to the limited pit space, in order to avoid crowding and trampling, security guards stopped at the door, feeling similar, two at a time to release in, until all released. Think of it as taking the subway to work in the morning, or going to the park in peak season

The test results

Squat No. :0Squat No. :1Squat No. :2Squat No. :3Squat No. :4Squat No. :5Squat No. :6Squat No. :7Process finished with exit code0
Copy the code

Semaphore’s constructor can pass whether a lock is fair or unfair, and the final test results are different. You can try it yourself.
When the test runs, it prints firstZero pit, one pit.Then 2 pits, 3 pits. Two, two releases every time. This is what Semaphore locks do.

3. Semaphore source code analysis

3.1 Constructors

public Semaphore(int permits) {
    sync = new NonfairSync(permits);
}

public Semaphore(int permits, boolean fair) {
    sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
Copy the code

An permitting, e.

By default, only the number of permitsis passed is required, that is, several threads are allowed at a time. The constructor creates an unfair lock. If you need to use a fair lock in a Semaphore shared lock, pass the second constructor argument fair = false/true. True: FairSync, fair lock. In our previous chapters, we have introduced the content and implementation of fair lock, as well as CLH and MCS fair Lock Introduction.

The initiallicenseThe number of

FairSync/NonfairSync(int permits) {
    super(permits);
}

Sync(int permits) {
    setState(permits);
}

protected final void setState(int newState) {
    state = newState;
}
Copy the code

When the constructor is initialized, the number of states in the AQS is set for both fair and unfair locks. This value is the value deducted and increased for the semaphore that can be obtained below.

3.2 Acquire semaphore

methods	describe
`semaphore.acquire()`	Get one semaphore at a time and respond to interrupt
`semaphore.acquire(2)`	N semaphores were obtained at a time, and the response was interrupted (2 pits at a time)
`semaphore.acquireUninterruptibly()`	Retrieves one semaphore at a time without responding to interrupts
`semaphore.acquireUninterruptibly(2)`	N semaphores are obtained at one time without interruption

In fact, the four methods to obtain the semaphore, is mainly, a few at a time and whether to respond to interrupt combination.
semaphore.acquire()The actual method called in the source code is, sync.acquireSharedInterruptibly(1). That’s the corresponding interrupt, one pit at a time.
semaphore.acquire(2)Similarly, this is going to be two at a time, which is the license.The scene in life is what I’ve arranged for my friend Yeah, here she comes. Come on in.

3.3 Acquire release semaphore

methods	describe
`semaphore.release()`	Release one semaphore at a time
`semaphore.release(2)`	Get n semaphores at a time

There must be a release to acquire a few semaphores to release a few semaphores. Acquire semaphore. Acquire (2) and release semaphore. Release (1)

3.4 Fair lock implementation

Semaphore acquisition process until fair lock is implemented. semaphore.acquire -> sync.acquireSharedInterruptibly(permits) -> tryAcquireShared(arg)

semaphore.acquire(1);

public void acquire(int permits) throws InterruptedException {
    if (permits < 0) throw new IllegalArgumentException();
    sync.acquireSharedInterruptibly(permits);
}

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}
Copy the code

FairSync.tryAcquireShared

protected int tryAcquireShared(int acquires) {
    for (;;) {
        if (hasQueuedPredecessors())
            return -1;
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            returnremaining; }}Copy the code

hasQueuedPredecessorsThe main implementation logic for fair locking lies in the use of this method. Its purpose is to determine whether a thread is ahead of it and to add a thread to the queue.We introduced implementations such as CLH earlier, which can be read in the previous section
for (;;), is a spin process that uses CAS to set the corresponding value of the state offset. In this way, we can avoid competing for semaphore conflicts in multiple threads.
getState()In the constructor, the value of state is initialized, where the semaphore is obtained using continuous CAS subtracting.
Note also that there is a difference between a shared lock and an exclusive lock. The exclusive lock returns true/false and the shared lock returns an int.
- If the value is less than 0, the current thread fails to acquire the shared lock.
- If this value is greater than 0, the current thread succeeded in acquiring the shared lock, and subsequent attempts by other threads to acquire the shared lock are likely to succeed.
- If this value is 0, the current thread succeeded in acquiring the shared lock, but subsequent attempts by other threads to acquire the shared lock fail.

3.5 Unfair Lock Implementation

NonfairSync.nonfairTryAcquireShared

protected int tryAcquireShared(int acquires) {
    return nonfairTryAcquireShared(acquires);
}

final int nonfairTryAcquireShared(int acquires) {
    for (;;) {
        int available = getState();
        int remaining = available - acquires;
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            returnremaining; }}Copy the code

With the realization of fair lock, the understanding of non-fair lock is relatively simple, just take itif (hasQueuedPredecessors())The judgment operation of.
Other logical implementations are consistent with fair locking.

3.6 Failed to obtain semaphore and was added to the synchronous waiting queue

We have seen the logic of normal semaphore acquisition in the implementation of fair and unfair locks. So what if you can’t get a semaphore normally at this point? In fact, this part of the thread needs to join the synchronization queue.

doAcquireSharedInterruptibly

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return; }}if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw newInterruptedException(); }}finally {
        if(failed) cancelAcquire(node); }}Copy the code

First of all,doAcquireSharedInterruptiblyThe method comes from the internal method of AQS, which is partly the same as when we learn competitive locking, but there are also some differences. Such as:addWaiter(Node.SHARED).tryAcquireSharedLet’s introduce this content.
Node.SHARED, actually has no special meaning, it is only a marker used to determine whether to share. `final boolean isShared() { return nextWaiter == SHARED;

} `

tryAcquireShared, mainly fromSemaphoreThe realization of fair lock and unfair lock in shared lock. Used to obtain synchronization status.
setHeadAndPropagate(node, r), if r > 0, the current thread node will be set as the head node after the synchronization is successful, and helpGC, p.ext = null, break the chain operation.
shouldParkAfterFailedAcquire(p, node), adjust the status of nodes in the synchronization queue and determine whether they should be suspended. This was covered in our previous article on locking.
parkAndCheckInterrupt()If the interrupt directly throws an exception, the request to the current node ends.
cancelAcquire(node)To cancel the thread request for this node.

4. CountDownLatch Uses the shared lock

CountDownLatch is also a type of shared lock, which is represented here because it is similar but different to Semaphore shared locks.

CountDownLatch is more about the idea of a group wave, again controlling the number of people, but needing a whole bunch. For example, we talked about four people kayaking together, two people on the seesaw together, two people squatting together. I’ve never seen that. That’s the idea of CountDownLatch.

public static void main(String[] args) throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(10);
    ExecutorService exec = Executors.newFixedThreadPool(10);
    for (int i = 0; i < 10; i++) {
        exec.execute(() -> {
            try {
                int millis = new Random().nextInt(10000);
                System.out.println("Waiting for tourists to board, time:" + millis + "(millis)");
                Thread.sleep(millis);
            } catch (Exception ignore) {
            } finally {
                latch.countDown(); // One quota will be deducted}}); }// Waiting for tourists
    latch.await();
    System.out.println("The captain's in a hurry, set sail!");
    // Close the thread pool
    exec.shutdown();
}
Copy the code

This is a park cruise scene case, waiting for 10 passengers to upload, they compare ink.
Subtract one from the last onelatch.countDown()
Wait for all the tourists to boardlatch.await()
Finally the captain sails!!irritable