| « | December 2025 | » | | 日 | 一 | 二 | 三 | 四 | 五 | 六 | | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 | 16 | 17 | 18 | 19 | 20 | 21 | 22 | 23 | 24 | 25 | 26 | 27 | 28 | 29 | 30 | 31 | | | | |
|
公告 |
| 本博客在此声明所有文章均为转摘，只做资料收集使用。并无其他商业用途。 |
| Blog信息 |
|
blog名称:
日志总数:210
评论数量:205
留言数量:-19
访问次数:930875
建立时间:2007年5月10日 |
| |
|
 [J2SE相关]JDK1.5新特性--java.util.concurrent BlockingQueue(3)
文章收藏, 网上资源, 软件技术, 电脑与网络
李小白 发表于 2007/7/18 17:45:29 |
|
lhwork 发表于 2006-12-11 11:15:33
并发库中的BlockingQueue是一个比较好玩的类，顾名思义，就是阻塞队列。该类主要提供了两个方法put()和take()，前者将一个对象放到队列中，如果队列已经满了，就等待直到有空闲节点；后者从head取一个对象，如果没有对象，就等待直到有可取的对象。
下面的例子比较简单，一个读线程，用于将要处理的文件对象添加到阻塞队列中，另外四个写线程用于取出文件对象，为了模拟写操作耗时长的特点，特让线程睡眠一段随机长度的时间。另外，该Demo也使用到了线程池和原子整型（AtomicInteger），AtomicInteger可以在并发情况下达到原子化更新，避免使用了synchronized，而且性能非常高。由于阻塞队列的put和take操作会阻塞，为了使线程退出，特在队列中添加了一个“标识”，算法中也叫“哨兵”，当发现这个哨兵后，写线程就退出。
当然线程池也要显式退出了。package concurrent;import java.io.File;import java.io.FileFilter;import java.util.concurrent.BlockingQueue;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.LinkedBlockingQueue;import java.util.concurrent.atomic.AtomicInteger;public class TestBlockingQueue { static long randomTime() { return (long) (Math.random() * 1000); } 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("F:\\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(); }} |
|
|