基于java中BlockingQueue的使用介紹

2019-11-26 16:10:57

字體：大中小

供稿：網(wǎng)友

最近在維護一個java工程，在群里面也就聊起來java的優(yōu)劣！無奈一些Java的終極粉絲，總是號稱性能已經(jīng)不必C++差，并且很多標準類庫都是大師級的人寫的，如何如何穩(wěn)定等等。索性就認真研究一番，他們給我的一項說明就是，在線程之間投遞消息，用java已經(jīng)封裝好的BlockingQueue，就足夠用了。

既然足夠用那就寫代碼測試嘍，簡簡單單寫一個小程序做了一番測試：

復(fù)制代碼代碼如下:

//默認包 
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(); 
            } 
        } 
    } 
} 

  
//需要添加的包 
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; 
}

傳遞十萬條數(shù)據(jù)，在我的測試機上面，大概需要50ms左右，倒是還可以！索性就看了一下BlockingQueue的底層實現(xiàn)

我在上面的測試代碼中使用的offer 和 poll，就看看這兩個實現(xiàn)函數(shù)吧，首先是offer

復(fù)制代碼代碼如下:

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

和一般的同步線程類似，只是多加了一個signal,在學習unix環(huán)境高級編程時候，看到條件變量用于線程之間的同步，可以實現(xiàn)線程以競爭的方式實現(xiàn)同步！
poll函數(shù)的實現(xiàn)也是類似！

復(fù)制代碼代碼如下:

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