快捷導(dǎo)航

基于java中BlockingQueue的使用介紹

更新時(shí)間：2013年04月19日 14:20:24 作者：

本篇文章小編為大家介紹，基于java中BlockingQueue的使用介紹。需要的朋友參考下

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

既然足夠用那就寫(xiě)代碼測(cè)試嘍，簡(jiǎn)簡(jiǎn)單單寫(xiě)一個(gè)小程序做了一番測(cè)試：

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

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

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

我在上面的測(cè)試代碼中使用的offer 和 poll，就看看這兩個(gè)實(shí)現(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; 
    }

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

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