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

This article introduces the knowledge of "how to understand the thread-safe queue ArrayBlockingQueue source code". In the operation of actual cases, many people will encounter such a dilemma, so let the editor lead you to learn how to deal with these situations. I hope you can read it carefully and be able to achieve something!

First, ArrayBlockingQueue source code analysis

ArrayBlockingQueue is a kind of queue, queue characteristics, first-out, first-out, but this kind of queue is a thread-safe blocking queue, why is it a blocking queue? I think this is exactly what I wrote and analyzed this article.

As this article involves a lot of content, so in some places I will not specifically talk about it in detail, but it is enough to make myself understand. Generally, when an article comes out, if it is difficult for me to read it, or if there is something I don't understand, I think, one kind of article is that the writer himself has omitted or sometimes vaguely written it.

However, I will not, because basically, my article is mainly written for myself. If I can help those in need, I am quite happy, because, as you may see, the style of the article I wrote before is different from that of others. I think I can write out my ideas at that time, if it's not perfect, it's okay, I can improve it later. I think this is what distinguishes me from other creators. I am not very interested in how much I read. Of course, if you follow me or share what I have written, I still thank you very much, . Let's analyze the source code of this queue collection.

Second, method analysis 2.0. You can slide around to view 2.1. The constructor / / must be given the capacity of the queue by default.

Public ArrayBlockingQueue (int capacity) {

This (capacity, false)

}

/ / step 2

Public ArrayBlockingQueue (int capacity, boolean fair) {

/ / of course, the capacity cannot be less than or equal to 0, because queues are used to load elements

/ / initialization capacity is 0, which is meaningless.

If (capacity 0) {

/ / when the number of elements in the queue increases, the putIndex value increases.

Final int putIndex = this.putIndex

Int I = takeIndex

Do {

/ / if the element o is equal to one of the elements of the array, false is returned directly.

If (o.equals (items [I]))

Return true

If (+ + I = = items.length)

I = 0

} while (I! = putIndex)

}

/ / if there are no elements in the queue, return false directly

Return false

} finally {

Lock.unlock ()

}

}

2.6 public E take take () method throws InterruptedException {

/ / Thread-safe methods

Final ReentrantLock lock = this.lock

Lock.lockInterruptibly ()

Try {

/ / if the number of elements in the queue is 0, you need to wait at this time, so this is a blocking queue.

/ / at this point, it is equivalent to an instruction that has been looping to determine whether the value of count is not equal to 0.

While (count = = 0)

NotEmpty.await ()

/ / perform out-queue operation

Return dequeue ()

} finally {

Lock.unlock ()

}

}

/ / step 2

Private E dequeue () {

/ / the default takeIndex starts at 0. If you leave the queue, the index value will be increased by one.

Final Object [] items = this.items

E x = (E) items [takeIndex]

/ / after leaving the queue, the element needs to be set to null and wait for gc to reclaim unreachable objects at some point.

Items [takeIndex] = null

/ / if takeIndex is equal to the size of the array space, it means that the number of elements in the queue has been taken, and you need to reset the index value to 0.

If (+ + takeIndex = = items.length)

TakeIndex = 0

/ / for each array element taken out, the number of elements is subtracted by one

Count--

If (itrs! = null)

Itrs.elementDequeued ()

/ / send a signal notification, "if the queue is not satisfied, you can also put the operation signal"

NotFull.signal ()

Return x

}

2.7 public E poll poll () method () {

/ / Thread safety

Final ReentrantLock lock = this.lock

Lock.lock ()

Try {

/ / if the number of elements in the queue is 0, then there are no elements in the queue, and null is returned.

/ / otherwise, the queue operation will be performed. It has been analyzed above, so we will not analyze it here.

Return (count = = 0)? Null: dequeue ()

} finally {

Lock.unlock ()

}

}

2.8 public void clear clear () method () {

Final Object [] items = this.items

Final ReentrantLock lock = this.lock

Lock.lock ()

Try {

Int k = count

/ / if there is an element greater than 0 in the queue, do the following directly

If (k > 0) {

/ / the location of putIndex is the location to which it needs to be moved.

Final int putIndex = this.putIndex

Int I = takeIndex

Do {

/ / Loop sets the value of each element to null and waits for gc to be triggered at some point

Items [I] = null

If (+ + I = = items.length)

I = 0

} while (I! = putIndex)

/ / Why are values assigned here? Think about it.

TakeIndex = putIndex

/ / the number of elements is set to 0

Count = 0

If (itrs! = null)

Itrs.queueIsEmpty ()

/ / notify that elements can be loaded in the queue.

For (; k > 0 & & lock.hasWaiters (notFull); KMel -)

NotFull.signal ()

}

} finally {

Lock.unlock ()

}

}

2.9 public String toString () () method () {

/ / Thread-safe methods

Final ReentrantLock lock = this.lock

Lock.lock ()

Try {

Int k = count

/ / if the number of elements in the queue is 0, "[]" is returned.

If (k = = 0)

Return "[]"

Final Object [] items = this.items

/ / use the StringBuilder method to splice each element of the queue

StringBuilder sb = new StringBuilder ()

Sb.append ('[')

For (int I = takeIndex;;) {

Object e = items [I]

Sb.append (e = = this? "(this Collection)": e)

If (--k = = 0)

Return sb.append (']') .toString ()

Sb.append (','). Append (')

If (+ + I = = items.length)

I = 0

}

} finally {

Lock.unlock ()

}

}

2.10 method public int remainingCapacity () {dint _ mainingCapacity ()

/ / get lock instance object

Final ReentrantLock lock = this.lock

Lock.lock ()

Try {

/ / the remaining space is equal to the size of the array space minus the size of the element is the size of the remaining space

Return items.length-count

} finally {

Lock.unlock ()

}

}

Third, summarize 3.1, inter-thread communication.

Based on the await () and singal () methods of Condition.

Import java.util.concurrent.ExecutorService

Import java.util.concurrent.Executors

Import java.util.concurrent.locks.Condition

Import java.util.concurrent.locks.Lock

Import java.util.concurrent.locks.ReentrantLock

/ * *

* @ author pc

, /

Public class ConditionTest {

Public Lock lock = new ReentrantLock ()

Public Condition condition = lock.newCondition ()

Public static void main (String [] args) {

ConditionTest test = new ConditionTest ()

ExecutorService executorService = Executors.newFixedThreadPool (2)

ExecutorService.execute ()-> test.conditionWait ()

ExecutorService.execute ()-> test.conditionSignal ()

}

Public void conditionWait () {

Lock.lock ()

Try {

System.out.println (Thread.currentThread () .getName () + "got the lock")

System.out.println (Thread.currentThread () .getName () + "wait for the signal")

Condition.await ()

System.out.println (Thread.currentThread (). GetName () + "get the signal")

} catch (Exception e) {

} finally {

Lock.unlock ()

}

}

Public void conditionSignal () {

Lock.lock ()

Try {

Thread.sleep (3000)

System.out.println (Thread.currentThread () .getName () + "got the lock")

Condition.signal ()

System.out.println (Thread.currentThread (). GetName () + "signal")

} catch (Exception e) {

} finally {

Lock.unlock ()

}

}

}

This is the end of "how to understand the thread-safe queue ArrayBlockingQueue source code". Thank you for reading. If you want to know more about the industry, you can follow the website, the editor will output more high-quality practical articles for you!

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.