Java生产者-消费者模式是多线程场景下协调生产线程和消费线程协作的经典设计模式，核心是通过共享缓冲区传递数据，但在实现过程中如果忽略线程安全细节，就会引发数据可见性和竞态条件问题，导致程序运行异常。
什么是数据可见性
数据可见性指的是当一个线程修改了共享变量的值，其他线程能够立即感知到这个修改。在Java内存模型中，每个线程有自己的工作内存，线程操作共享变量时会先把变量从主内存拷贝到工作内存，修改后再写回主内存。如果变量没有正确处理，线程可能一直使用工作内存中的旧值，看不到其他线程的修改。
在生产者-消费者模式中，共享缓冲区的状态变量如果没有保证可见性，消费线程可能不知道缓冲区已经有新数据，或者生产线程不知道缓冲区已经满了，导致程序逻辑出错。
什么是竞态条件
竞态条件指的是程序的执行结果依赖于多个线程交替执行的时序，当多个线程同时操作共享资源，并且操作顺序会影响最终结果时，就会产生竞态条件。在生产者-消费者模式中，常见的竞态条件场景是多个生产线程同时往缓冲区写数据，或者多个消费线程同时取数据，导致数据覆盖、重复消费或者下标越界等问题。
问题示例分析
下面是一段没有处理可见性和竞态条件的生产者-消费者实现代码，我们来看看问题出在哪里：
// 共享缓冲区，没有做线程安全处理
class SharedBuffer {
private int[] data = new int[5];
private int count = 0; // 当前缓冲区数据数量
private int putIndex = 0; // 生产下标
private int takeIndex = 0; // 消费下标
// 生产方法
public void put(int value) {
// 没有同步，多个线程同时进入会出问题
if (count < data.length) {
data[putIndex] = value;
putIndex = (putIndex + 1) % data.length;
count++;
System.out.println(Thread.currentThread().getName() + " 生产了数据：" + value + "，当前数量：" + count);
} else {
System.out.println(Thread.currentThread().getName() + " 缓冲区已满，无法生产");
}
}
// 消费方法
public int take() {
// 没有同步，多个线程同时进入会出问题
if (count > 0) {
int value = data[takeIndex];
takeIndex = (takeIndex + 1) % data.length;
count--;
System.out.println(Thread.currentThread().getName() + " 消费了数据：" + value + "，当前数量：" + count);
return value;
} else {
System.out.println(Thread.currentThread().getName() + " 缓冲区为空，无法消费");
return -1;
}
}
}
// 生产者线程
class Producer extends Thread {
private SharedBuffer buffer;
public Producer(SharedBuffer buffer) {
this.buffer = buffer;
}
@Override
public void run() {
for (int i = 0; i < 10; i++) {
buffer.put(i);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
// 消费者线程
class Consumer extends Thread {
private SharedBuffer buffer;
public Consumer(SharedBuffer buffer) {
this.buffer = buffer;
}
@Override
public void run() {
for (int i = 0; i < 10; i++) {
buffer.take();
try {
Thread.sleep(150);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public class Test {
public static void main(String[] args) {
SharedBuffer buffer = new SharedBuffer();
// 启动2个生产者
new Producer(buffer).start();
new Producer(buffer).start();
// 启动2个消费者
new Consumer(buffer).start();
new Consumer(buffer).start();
}
}
这段代码的问题主要有两点：第一，count、putIndex、takeIndex这些共享变量没有保证可见性，线程修改后其他线程可能看不到最新值；第二，put和take方法没有加锁，多个线程同时执行时会产生竞态条件，比如两个生产线程同时判断count < data.length为真，然后同时写数据，会导致数据覆盖，或者count的值计算错误。
解决方案
1. 使用synchronized保证原子性和可见性
synchronized关键字可以保证同一时间只有一个线程执行同步代码块，同时保证修改后的变量会立即刷新到主内存，其他线程进入同步块时会从主内存重新读取变量，解决可见性问题，同时把操作变成原子操作，避免竞态条件。
class SafeSharedBuffer {
private int[] data = new int[5];
private int count = 0;
private int putIndex = 0;
private int takeIndex = 0;
// 生产方法，加synchronized保证同步
public synchronized void put(int value) {
// 循环检查，避免虚假唤醒
while (count == data.length) {
try {
// 缓冲区满，等待
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
data[putIndex] = value;
putIndex = (putIndex + 1) % data.length;
count++;
System.out.println(Thread.currentThread().getName() + " 生产了数据：" + value + "，当前数量：" + count);
// 唤醒等待的消费者线程
notifyAll();
}
// 消费方法，加synchronized保证同步
public synchronized int take() {
// 循环检查，避免虚假唤醒
while (count == 0) {
try {
// 缓冲区空，等待
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
int value = data[takeIndex];
takeIndex = (takeIndex + 1) % data.length;
count--;
System.out.println(Thread.currentThread().getName() + " 消费了数据：" + value + "，当前数量：" + count);
// 唤醒等待的生产者线程
notifyAll();
return value;
}
}
2. 使用Lock和Condition实现更灵活的控制
除了synchronized，还可以使用java.util.concurrent.locks.Lock和Condition，可以创建多个等待队列，分别控制生产者和消费者的等待唤醒，效率更高。
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
class LockSharedBuffer {
private int[] data = new int[5];
private int count = 0;
private int putIndex = 0;
private int takeIndex = 0;
private Lock lock = new ReentrantLock();
// 生产者等待条件：缓冲区满
private Condition notFull = lock.newCondition();
// 消费者等待条件：缓冲区空
private Condition notEmpty = lock.newCondition();
public void put(int value) {
lock.lock();
try {
while (count == data.length) {
// 缓冲区满，生产者等待
notFull.await();
}
data[putIndex] = value;
putIndex = (putIndex + 1) % data.length;
count++;
System.out.println(Thread.currentThread().getName() + " 生产了数据：" + value + "，当前数量：" + count);
// 唤醒等待的消费者
notEmpty.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public int take() {
lock.lock();
try {
while (count == 0) {
// 缓冲区空，消费者等待
notEmpty.await();
}
int value = data[takeIndex];
takeIndex = (takeIndex + 1) % data.length;
count--;
System.out.println(Thread.currentThread().getName() + " 消费了数据：" + value + "，当前数量：" + count);
// 唤醒等待的生产者
notFull.signal();
return value;
} catch (InterruptedException e) {
e.printStackTrace();
return -1;
} finally {
lock.unlock();
}
}
}
3. 使用并发容器简化实现
Java并发包提供了现成的线程安全容器，比如ArrayBlockingQueue，内部已经实现了生产者-消费者模式需要的同步和等待唤醒逻辑，直接使用可以避免自己处理可见性和竞态条件问题。
import java.util.concurrent.ArrayBlockingQueue;
class QueueSharedBuffer {
// 创建容量为5的阻塞队列
private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);
public void put(int value) {
try {
// 队列满时会自动阻塞
queue.put(value);
System.out.println(Thread.currentThread().getName() + " 生产了数据：" + value + "，当前数量：" + queue.size());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public int take() {
try {
// 队列空时会自动阻塞
int value = queue.take();
System.out.println(Thread.currentThread().getName() + " 消费了数据：" + value + "，当前数量：" + queue.size());
return value;
} catch (InterruptedException e) {
e.printStackTrace();
return -1;
}
}
}
总结
在Java生产者-消费者模式中，数据可见性问题的核心是共享变量没有及时在主内存和工作内存之间同步，竞态条件的核心是多个线程同时操作共享资源且操作不是原子性的。解决这类问题的关键是保证共享资源的操作原子性，同时保证变量的可见性，可以通过synchronized、Lock或者直接使用并发容器来实现。实际开发中如果不需要特别定制逻辑，优先选择ArrayBlockingQueue这类成熟的并发容器，能减少出错概率。