首页 > 代码库 > 队列 LinkedBlockingQueue

队列 LinkedBlockingQueue

2024-10-12 09:22:39   216人阅读

1 api

    java.util.concurrent包下的新类。LinkedBlockingQueue就是其中之一，是一个阻塞的线程安全的队列，底层采用链表实现。
     
      LinkedBlockingQueue构造的时候若没有指定大小，则默认大小为Integer.MAX_VALUE，当然也可以在构造函数的参数中指定大小。LinkedBlockingQueue不接受null。

    添加元素的方法有三个：add,put,offer,且这三个元素都是向队列尾部添加元素的意思。

    区别:

        add方法在添加元素的时候，若超出了度列的长度会直接抛出异常： 

      put方法，若向队尾添加元素的时候发现队列已经满了会发生阻塞一直等待空间，以加入元素。 
    offer方法在添加元素时，如果发现队列已满无法添加的话，会直接返回false。     
    
    从队列中取出并移除头元素的方法有：poll，remove，take。     

        poll: 若队列为空，返回null。

        remove:若队列为空，抛出NoSuchElementException异常。

        take:若队列为空，发生阻塞，等待有元素。

2基于LinkedBlockingQueue的生产者和消费者

3示例2

 并发库中的BlockingQueue是一个比较好玩的类，顾名思义，就是阻塞队列。该类主要提供了两个方法put()和take()，前者将一个对象放到队列中，如果队列已经满了，就等待直到有空闲节点；后者从head取一个对象，如果没有对象，就等待直到有可取的对象。

 下面的例子比较简单，一个读线程，用于将要处理的文件对象添加到阻塞队列中，
 另外四个写线程用于取出文件对象，为了模拟写操作耗时长的特点，特让线程睡眠一段随机长度的时间。另外，该Demo也使用到了线程池和原子整型（AtomicInteger），AtomicInteger可以在并发情况下达到原子化更新，避免使用了synchronized，而且性能非常高。由于阻塞队列的put和take操作会阻塞，为了使线程退出，特在队列中添加了一个“标识”，算法中也叫“哨兵”，当发现这个哨兵后，写线程就退出。


public class LinkedBlockingQueueTest {
    static long randomTime() {
        return (long) (Math.random() * 1000);
    }
    @Test
    public void testName() throws Exception {
        AtomicInteger rc = new AtomicInteger();
        int incrementAndGet = rc.incrementAndGet();
        System.out.println(incrementAndGet);
    }
    
    
    public static void main(String[] args) {
        // 能容纳100个文件
        final BlockingQueue<File> queue = new LinkedBlockingQueue<File>(100);
        // 线程池
        final ExecutorService exec = Executors.newFixedThreadPool(5);
        final File root = new File("D:\\JavaLib");
        // 完成标志
        final File exitFile = new File("");
        // 读个数
        final AtomicInteger rc = new AtomicInteger();
        // 写个数
        final AtomicInteger wc = new AtomicInteger();
        // 读线程
        Runnable read = new Runnable() {
            public void run() {
                scanFile(root);
                scanFile(exitFile);
            }
            public void scanFile(File file) {
                if (file.isDirectory()) {
                    File[] files = file.listFiles(new FileFilter() {
                        public boolean accept(File pathname) {
                            return pathname.isDirectory() || pathname.getPath().endsWith(".java");
                        }
                    });
                    for (File one : files)
                        scanFile(one);
                } else {
                    try {
                        int index = rc.incrementAndGet();
                        System.out.println("Read0: " + index + " " + file.getPath());
                        queue.put(file);
                    } catch (InterruptedException e) {
                    }
                }
            }
        };
        exec.submit(read);
        // 四个写线程
        for (int index = 0; index < 4; index++) {
            // write thread
            final int NO = index;
            Runnable write = new Runnable() {
                String threadName = "Write" + NO;
                public void run() {
                    while (true) {
                        try {
                            Thread.sleep(randomTime());
                            int index = wc.incrementAndGet();
                            File file = queue.take();
                            // 队列已经无对象
                            if (file == exitFile) {
                                // 再次添加"标志"，以让其他线程正常退出
                                queue.put(exitFile);
                                break;
                            }
                            System.out.println(threadName + ": " + index + " " + file.getPath());
                        } catch (InterruptedException e) {
                        }
                    }
                }
            };
            exec.submit(write);
        }
        exec.shutdown();
    }
}

















null


队列 LinkedBlockingQueue


联系
我们