Java并发编程：LinkedBlockingQueue的实际应用

在Java并发编程中，BlockingQueue是一个非常重要的工具，特别适用于多线程环境下需要协调生产者和消费者之间数据传递的场景。它提供了线程安全的操作，使得一个线程可以将数据放入队列中，而另一个线程可以从队列中取出数据，即使在高并发的情况下也能保证数据的一致性和安全性。

### LinkedBlockingQueue简介

`LinkedBlockingQueue`是BlockingQueue接口的一个实现，内部使用链表结构存储元素，默认情况下容量无界，但也可以指定最大容量。它的主要特点包括：
- **put()方法**：将元素插入队列尾部。如果队列已满，则阻塞当前线程直到有空闲空间。
- **take()方法**：从队列头部移除并返回元素。如果队列为空，则阻塞当前线程直到有可用元素。

### 示例代码解析

下面是一个简单的示例程序，演示了如何使用`LinkedBlockingQueue`来处理文件读取和写入任务。该程序创建了一个读线程用于将待处理的文件对象添加到阻塞队列中，同时启动四个写线程用于从队列中取出文件对象进行处理。为了模拟实际的写操作耗时较长的情况，每个写线程在处理完文件后会随机休眠一段时间。

此外，程序还使用了线程池（`ExecutorService`）来管理线程的生命周期，并引入了原子类`AtomicInteger`以确保计数器的线程安全性，从而避免了显式的同步机制。

#### 关键点说明

1. **线程退出机制**：为了确保所有写线程能够正常退出，我们在队列中加入了一个特殊的“哨兵”对象（即一个空文件）。当某个写线程遇到这个哨兵时，它会再次将其放回队列，以便其他写线程也能检测到并退出。
2. **线程池管理**：使用`Executors.newFixedThreadPool()`创建固定大小的线程池，简化了线程的创建和销毁过程。
3. **原子类的应用**：`AtomicInteger`提供了高效的原子操作，能够在不使用锁的情况下完成计数器的递增和获取。

```java
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 queue = new LinkedBlockingQueue<>(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 readTask = new Runnable() {
public void run() {
scanFile(root);
scanFile(exitFile); // 添加哨兵
}

private 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("Read: " + index + " " + file.getPath());
queue.put(file);
} catch (InterruptedException e) {}
}
}
};

exec.submit(readTask);

// 定义写线程任务
for (int i = 0; i <4; i++) {
final int NO = i;
Runnable writeTask = 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(writeTask);
}

exec.shutdown();
}
}
```

通过以上代码，我们可以看到如何利用`LinkedBlockingQueue`实现生产者-消费者模式，并结合线程池和原子类提高程序的效率和稳定性。