引言
在高并发系统中，生产者-消费者模式是最经典的设计模式之一。当生产者和消费者速度不匹配时，需要一个缓冲区来解耦两者。普通的数组队列在出队时需要O(n)的元素移动，而链表队列虽然入队出队都是O(1)，但每个元素都有额外的指针开销。
环形缓冲区（Circular Buffer / Ring Buffer）是一种固定大小的循环数组结构，利用模运算实现O(1)的入队和出队，同时具有极高的缓存局部性。本文将用Java实现一个线程安全的环形缓冲区，并展示其在流式数据处理中的应用。
核心原理
环形缓冲区的核心思想是将线性数组的首尾相连，通过两个指针（或索引）管理数据：
- head（头指针）：指向下一个要读取的位置
- tail（尾指针）：指向下一个要写入的位置
关键操作：
– 入队（enqueue）：在tail位置写入数据，然后 tail = (tail + 1) % capacity
– 出队（dequeue）：从head位置读取数据，然后 head = (head + 1) % capacity
– 判满：(tail + 1) % capacity == head（牺牲一个空间区分空和满）
– 判空：head == tail
Java完整实现
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
* 线程安全环形缓冲区（阻塞队列）
* 适用于生产者-消费者模式的流式数据处理
* @param <T> 元素类型
*/
public class CircularBuffer<T> {
private final Object[] buffer; // 存储数组
private final int capacity; // 容量（实际可用为capacity-1）
private int head; // 头指针（读取位置）
private int tail; // 尾指针（写入位置）
private int count; // 当前元素数量
// 线程同步机制
private final ReentrantLock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
public CircularBuffer(int capacity) {
if (capacity <= 0) {
throw new IllegalArgumentException("容量必须大于0");
}
this.capacity = capacity;
this.buffer = new Object[capacity];
this.head = 0;
this.tail = 0;
this.count = 0;
}
/**
* 阻塞式入队
* 当缓冲区满时等待，直到有空间可用
*/
public void put(T item) throws InterruptedException {
lock.lock();
try {
// 等待直到缓冲区不满
while (count == capacity) {
notFull.await();
}
buffer[tail] = item;
tail = (tail + 1) % capacity;
count++;
// 通知等待的消费者
notEmpty.signal();
} finally {
lock.lock.unlock();
}
}
/**
* 阻塞式出队
* 当缓冲区空时等待，直到有数据可读
*/
@SuppressWarnings("unchecked")
public T take() throws InterruptedException {
lock.lock();
try {
// 等待直到缓冲区不空
while (count == 0) {
notEmpty.await();
}
T item = (T) buffer[head];
buffer[head] = null; // 帮助GC
head = (head + 1) % capacity;
count--;
// 通知等待的生产者
notFull.signal();
return item;
} finally {
lock.unlock();
}
}
/**
* 非阻塞入队，返回是否成功
*/
public boolean offer(T item) {
lock.lock();
try {
if (count == capacity) {
return false; // 缓冲区已满
}
buffer[tail] = item;
tail = (tail + 1) % capacity;
count++;
notEmpty.signal();
return true;
} finally {
lock.unlock();
}
}
/**
* 非阻塞出队，返回元素或null
*/
@SuppressWarnings("unchecked")
public T poll() {
lock.lock();
try {
if (count == 0) {
return null; // 缓冲区为空
}
T item = (T) buffer[head];
buffer[head] = null;
head = (head + 1) % capacity;
count--;
notFull.signal();
return item;
} finally {
lock.unlock();
}
}
/**
* 查看队首元素（不移除）
*/
@SuppressWarnings("unchecked")
public T peek() {
lock.lock();
try {
if (count == 0) {
return null;
}
return (T) buffer[head];
} finally {
lock.unlock();
}
}
public int size() {
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}
public boolean isEmpty() {
return size() == 0;
}
public boolean isFull() {
lock.lock();
try {
return count == capacity;
} finally {
lock.unlock();
}
}
/**
* 清空缓冲区
*/
public void clear() {
lock.lock();
try {
for (int i = 0; i < capacity; i++) {
buffer[i] = null;
}
head = tail = count = 0;
notFull.signalAll();
} finally {
lock.unlock();
}
}
@Override
public String toString() {
lock.lock();
try {
StringBuilder sb = new StringBuilder("[");
for (int i = 0; i < count; i++) {
int idx = (head + i) % capacity;
sb.append(buffer[idx]);
if (i < count - 1) sb.append(", ");
}
sb.append("]");
return sb.toString();
} finally {
lock.unlock();
}
}
// ==================== 流式数据处理应用示例 ====================
/**
* 日志收集器的流式处理示例
*/
static class LogCollector {
private final CircularBuffer<String> logBuffer;
private volatile boolean running = true;
LogCollector(int bufferSize) {
this.logBuffer = new CircularBuffer<>(bufferSize);
}
// 生产者：接收日志
void produceLog(String log) {
try {
logBuffer.put(log);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
// 消费者：批量处理日志
void startConsumer() {
new Thread(() -> {
while (running || !logBuffer.isEmpty()) {
try {
String log = logBuffer.take();
processLog(log);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
}
}
}, "LogConsumer").start();
}
void processLog(String log) {
System.out.println("[处理] " + log);
}
void shutdown() {
running = false;
}
}
/**
* 滑动窗口平均值计算（流式统计）
*/
static class SlidingWindowAverage {
private final CircularBuffer<Double> window;
private double sum = 0;
SlidingWindowAverage(int windowSize) {
this.window = new CircularBuffer<>(windowSize);
}
void add(double value) {
// 如果窗口已满，移除最旧的值
if (window.isFull()) {
Double old = window.poll();
if (old != null) sum -= old;
}
window.offer(value);
sum += value;
}
double getAverage() {
int size = window.size();
return size == 0 ? 0 : sum / size;
}
}
public static void main(String[] args) throws InterruptedException {
System.out.println("=== 环形缓冲区基础测试 ===");
CircularBuffer<Integer> cb = new CircularBuffer<>(5);
// 测试入队
for (int i = 1; i <= 3; i++) {
cb.put(i);
System.out.println("入队 " + i + ", 当前: " + cb);
}
// 测试出队
System.out.println("出队: " + cb.take());
System.out.println("出队后: " + cb);
// 测试环绕
cb.put(4);
cb.put(5);
cb.put(6); // 此时应环绕到数组开头
System.out.println("环绕后: " + cb);
System.out.println("\n=== 滑动窗口平均值测试 ===");
SlidingWindowAverage swa = new SlidingWindowAverage(3);
double[] data = {10, 20, 30, 40, 50};
for (double d : data) {
swa.add(d);
System.out.printf("添加 %.0f, 窗口平均值: %.2f%n", d, swa.getAverage());
}
System.out.println("\n=== 多线程生产者-消费者测试 ===");
CircularBuffer<Integer> queue = new CircularBuffer<>(10);
// 生产者线程
Thread producer = new Thread(() -> {
for (int i = 0; i < 20; i++) {
try {
queue.put(i);
System.out.println("生产者生产: " + i);
Thread.sleep(50);
} catch (InterruptedException e) {
break;
}
}
});
// 消费者线程
Thread consumer = new Thread(() -> {
for (int i = 0; i < 20; i++) {
try {
int val = queue.take();
System.out.println(" 消费者消费: " + val);
Thread.sleep(100);
} catch (InterruptedException e) {
break;
}
}
});
producer.start();
consumer.start();
producer.join();
consumer.join();
System.out.println("\n所有测试完成!");
}
}
复杂度分析
| 操作 | 时间复杂度 | 空间复杂度 | 说明 |
|---|---|---|---|
| 入队（put/offer） | O(1) | O(1) | 模运算实现循环 |
| 出队（take/poll） | O(1) | O(1) | 直接读取head位置 |
| 查看队首（peek） | O(1) | O(1) | 不修改指针 |
| 整体结构 | – | O(capacity) | 预分配固定数组 |
与链表队列相比，环形缓冲区没有节点对象的开销，数据紧密排列在数组中，具有更好的CPU缓存局部性（Cache Locality），在高吞吐场景下性能提升显著。
应用场景
- 网络数据包缓冲：网卡驱动与协议栈之间的数据传递。
- 音频/视频流处理：实时音视频采集与编码之间的缓冲。
- 日志收集系统：高并发日志写入与磁盘持久化之间的解耦。
- 嵌入式系统：内存受限环境下的高效队列实现。
总结
环形缓冲区是计算机科学中最优雅的数据结构之一。它将数组的首尾通过模运算无缝连接，在固定空间内实现了高效的FIFO队列。配合条件变量实现的阻塞语义，它构成了生产者-消费者模式的坚实基石。理解环形缓冲区，对于掌握并发编程和系统优化具有重要意义。
