Introduction to the use of BlockingQueue based on java

Author：Eve Cole Update Time：2024-11-18 14:48:02

Recently I am maintaining a Java project, and in the group we are chatting about the pros and cons of Java! Unfortunately, some ultimate fans of Java always claim that performance is no longer as bad as C++, and many standard libraries are written by masters, how stable they are, etc. I simply studied it carefully, and one of the instructions they gave me was that to deliver messages between threads, it is enough to use BlockingQueue, which has been encapsulated in java.

Since it is enough, let’s write code to test it. Simply write a small program and do some testing:

Copy the code code as follows:

//Default package
import java.util.concurrent.*;

import base.MyRunnable;

public class Test
{
public static void main(String[] args)
{
BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();
java.lang.Runnable r = new MyRunnable(queue);
Thread t = new Thread(r);
t.start();

while(true)
{
try
{
while(true)
{
for(int i =0;i < 10000;i++)
{
queue.offer(i);
}
}
}
catch (Exception e)
{
e.printStackTrace();
}
}
}
}

//Packages that need to be added
package base;

import java.lang.Runnable;
import java.util.concurrent.*;
import java.util.*;

public class MyRunnable implements Runnable
{
public MyRunnable(BlockingQueue<Integer> queue)
{
this.queue = queue;
}
public void run()
{
Date d = new Date();
long starttime = d.getTime();
System.err.println(starttime);
int count = 0;
while(true)
{
try
{
Integer i = this.queue.poll();
if(i != null)
{
count++;
}
if(count == 100000)
{
Date e = new Date();
long endtime = e.getTime();
System.err.println(count);
System.err.println(endtime);
System.err.print(endtime - starttime);
break;
}

}
catch (Exception e)
{

}
}
}
private BlockingQueue<Integer> queue;
}

To transfer 100,000 pieces of data, it takes about 50ms on my test machine, which is not bad! Just take a look at the underlying implementation of BlockingQueue

I used offer and poll in the above test code. Let’s take a look at these two implementation functions. The first is offer

Copy the code code as follows:

public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
E x = null;
int c = -1;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
if (count.get() > 0) {
x = extract();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
}
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}

It is similar to a general synchronization thread, except that one more signal is added. When learning advanced programming in the Unix environment, I saw that condition variables are used for synchronization between threads, and threads can be synchronized in a competitive manner!
The implementation of the poll function is similar!

Copy the code code as follows:

public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false;
int c = -1;
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
if (count.get() < capacity) {
insert(e);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
}
} finally {
putLock.unlock();
}
if(c==0)
signalNotEmpty();
return c >= 0;
}