Shulou(Shulou.com)06/03 Report--

This article mainly explains "how to use AQS shared lock, Semaphore, CountDownLatch". The content of the article is simple and clear, easy to learn and understand. Please follow the editor's train of thought to study and learn "how to use AQS shared lock, Semaphore, CountDownLatch".

Shared lock and AQS1. What are the locks implemented based on AQS? Figure 18-1 locking based on AQS

AQS (AbstractQueuedSynchronizer) is a very important class in the Java concurrent package, and most of the locks are implemented based on AQS, including:

ReentrantLock, reentrant lock. This is the lock we first introduced and the most commonly used lock. It is usually compared with synchronized. ReentrantReadWriteLock, read-write lock. The read lock is a shared lock and the write lock is an exclusive lock. Semaphore, semaphore lock. Mainly used to control traffic, for example: database connection pool to assign you 10 links, then let you connect to one, even 10 has not been released, then you and so on. CountDownLatch, lock up. Latch latch means, for example, a rafting boat for four people, push it into the water when it is full.

In this chapter, we mainly introduce Semaphore, the implementation of semaphore lock, which is about the use of shared lock and source code analysis.

2. Semaphore shared lock uses Semaphore semaphore = new Semaphore (2, false); / / input parameter of constructor, permits: semaphore, fair: fair lock / unfair lock

For (int I = 0; I

< 8; i++) { new Thread(() ->

{

Try {

Semaphore.acquire ()

System.out.println (Thread.currentThread (). GetName () + "squatting pit")

Thread.sleep (1000L)

} catch (InterruptedException ignore) {

} finally {

Semaphore.release ()

}

}, "squat pit number:" + I) .start ()

}

Here we simulate a scene of squatting in a toilet in a high-speed service area. Due to the limited pit space, in order to avoid crowding and trampling, the security guards stopped at the door and felt almost the same, releasing two at a time until both were released. You can also think of going to work by subway in the morning or going to the park in the peak season.

"Test results"

Squat pit number: 0 squat pit

Squat pit number: 1 squat pit

Squatting pit number: 2 squatting pit

Squat pit number: 3 squat pit

Squat pit number: 4 squat pit

Squat pit number: 5 squat pit

Squat pit number: 6 squat pit

Squat pit number: 7 squat pit

Process finished with exit code 0

The constructor of Semaphore can pass whether a fair lock or an unfair lock, and the final test results are different, so you can try it on your own. When the test is running, it will first output 0 pit, 1 pit, and then 2 pit, 3 pit. Each time it is the release of two, two. This is what Semaphore signal quantity locks do. 3. Semaphore source code analysis 3.1 Constructor public Semaphore (int permits) {

Sync = new NonfairSync (permits)

}

Public Semaphore (int permits, boolean fair) {

Sync = fair? New FairSync (permits): new NonfairSync (permits)

}

Permits:n. A license or concession, esp of a limited period.

By default, you only need to pass in the number of permits licenses, which allows you to release several threads at a time. The constructor creates an unfair lock. If you need to use the fair lock in the Semaphore shared lock, you can pass in the parameter fair = false/true of the second constructor. True:FairSync, fair lock. In our previous chapters, we have introduced the related contents and implementation of fair lock, as well as the introduction of fair lock by CLH and MCS.

"initial license quantity"

FairSync/NonfairSync (int permits) {

Super (permits)

}

Sync (int permits) {

SetState (permits)

}

Protected final void setState (int newState) {

State = newState

}

When the constructor is initialized, the value of state in AQS is set, regardless of whether it is a fair lock or an unfair lock. This value is for the semaphore deductions and increments that can be obtained below.

3.2 the acquire acquisition semaphore method describes that semaphore.acquire () acquires one semaphore at a time, response interrupt semaphore.acquire (2) obtains n semaphores at a time, responds to interrupts (2 pits at a time) semaphore.acquireUninterruptibly () acquires one semaphore at a time, does not respond to interrupt semaphore.acquireUninterruptibly (2) acquires n semaphores at a time, and actually obtains semaphores without responding to interrupts. Gets a combination of several at a time and whether to respond to interrupts. Semaphore.acquire (), the method actually called in the source code is sync.acquireSharedInterruptibly (1). That is, the corresponding interruption, accounting for only one pit at a time. Semaphore.acquire (2), similarly, this is to take up two places at a time, that is, the license. The scene in life is that I made the right arrangement for my friend. Here she comes. Come on in. Description of acquire release semaphore method semaphore.release () releases one semaphore at a time semaphore.release (2) acquires n semaphores at a time

Where there is acquisition, there must be release, and if several semaphores are acquired, several semaphores will be released. Of course, you can try to get two semaphores semaphore.acquire (2), release semaphores semaphore.release (1), and see how it works.

3.4 implementation of fair lock

"semaphore acquisition process" until the fair lock is realized. 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); } 「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)) return remaining; } } hasQueuedPredecessors，公平锁的主要实现逻辑都在于这个方法的使用。它的目的就是判断有线程排在自己前面没，以及把线程添加到队列中的逻辑实现。 在前面我们介绍过CLH等实现，可以往前一章节阅读for (;;)，是一个自旋的过程，通过 CAS 来设置 state 偏移量对应值。这样就可以避免多线程下竞争获取信号量冲突。getState()，在构造函数中已经初始化 state 值，在这里获取信号量时就是使用 CAS 不断的扣减。另外需要注意，共享锁和独占锁在这里是有区别的，独占锁直接返回true/false，共享锁返回的是int值。如果该值小于0，则当前线程获取共享锁失败。如果该值大于0，则当前线程获取共享锁成功，并且接下来其他线程尝试获取共享锁的行为很可能成功。如果该值等于0，则当前线程获取共享锁成功，但是接下来其他线程尝试获取共享锁的行为会失败。3.5 非公平锁实现 「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)) return remaining; } } 有了公平锁的实现，非公平锁的理解就比较简单了，只是拿去了 if (hasQueuedPredecessors()) 的判断操作。其他的逻辑实现都和公平锁一致。3.6 获取信号量失败，加入同步等待队列 在公平锁和非公平锁的实现中，我们已经看到正常获取信号量的逻辑。那么如果此时不能正常获取信号量呢？其实这部分线程就需要加入到同步队列。 「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 new InterruptedException ()

}

} finally {

If (failed)

CancelAcquire (node)

}

}

First of all, the doAcquireSharedInterruptibly method comes from the internal method of AQS, which is the same as when we learn competitive locking, but there are also some differences. For example: addWaiter (Node.SHARED), tryAcquireShared, let's mainly introduce this content. Node.SHARED, in fact, has no special meaning, it is just a marking function to determine whether it is shared or not. Final boolean isShared () {return nextWaiter = = SHARED;} tryAcquireShared, mainly from the implementation of fair locks and unfair locks in Semaphore shared locks. Used to get the synchronization status. SetHeadAndPropagate (node, r), if r > 0, the current thread node is set to the header node after the synchronization is successful, and helpGC,p.next = null, chain breaking operation. ShouldParkAfterFailedAcquire (p, node) adjusts the state of the node node in the synchronization queue and determines whether it should be suspended. This has been introduced in our previous article on locks. ParkAndCheckInterrupt () to determine whether it needs to be interrupted, and if the interrupt throws an exception directly, the current node request ends. CancelAcquire (node), which cancels the thread request for this node. 4. Use of CountDownLatch shared locks

CountDownLatch is also a type of shared lock, which is reflected here because it is similar and different from Semaphore shared lock.

CountDownLatch embodies more of the idea of forming a group, which is also about controlling the number of people, but it needs to be full of people. For example, I have never seen four people kayaking together, two people on the seesaw together, and two people squatting together. This is the idea of latch CountDownLatch lock.

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-consuming:" + millis + "(millis)")

Thread.sleep (millis)

} catch (Exception ignore) {

} finally {

Latch.countDown (); / / deduction of one quota at a time

}

});

}

/ / waiting for tourists

Latch.await ()

System.out.println ("the captain is impatient, set sail!")

/ / close the thread pool

Exec.shutdown ()

}

In this case of a park cruise, waiting for 10 passengers to upload, they compare ink. Last deduction a latch.countDown () waiting for tourists to get on board latch.await () finally captain set sail! Impatient

"Test results"

Waiting for tourists to board, time: 6689 (millis)

Waiting for tourists to board, time: 2303 (millis)

Waiting for tourists to board, time: 8208 (millis)

Waiting for tourists to board, time: 435 (millis)

Waiting for tourists to board, time: 9489 (millis)

Waiting for tourists to board, time: 4937 (millis)

Waiting for tourists to board, time: 2771 (millis)

Waiting for tourists to board, time: 4823 (millis)

Waiting for tourists to board, time: 1989 (millis)

Waiting for tourists to board, time: 8506 (millis)

The captain is impatient, set sail!

Process finished with exit code 0

In your actual test, you will find that the captain is impatient and will have to wait for some time to sail. What is reflected here is the idea of the latch, forming a team and taking it away in a wave. The implementation of CountDownLatch is basically the same as Semaphore, but the details are slightly different, so there is no longer source code analysis. Thank you for reading, the above is "how to use AQS shared lock, Semaphore, CountDownLatch" content, after the study of this article, I believe you on how to use AQS shared lock, Semaphore, CountDownLatch this problem has a deeper understanding, the specific use of the situation also needs to be verified in practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.