当前位置：首页 > 技术 > 正文内容

Java生产者-消费者模式实现详解

访客技术 2026年10月26日 1

在现代软件开发中，处理高并发场景是常态。大型电商平台如淘宝，为了在数秒内响应用户请求，其背后运行着成千上万台服务器，并采用复杂的路由和资源分发机制。而服务间的解耦和通信，则常常依赖于消息队列，这本质上就是生产者-消费者模式的应用。

现实生活中的类比我们日常生活中，肯德基和麦当劳的点餐流程可以很好地说明两种不同的处理模式：

肯德基模式： 顾客点餐后，服务员一次性准备好所有餐品并交付，然后收款。这类似于某些同步处理方式，一个线程（服务员）处理完所有逻辑。
麦当劳模式： 顾客点餐后立即付款，然后到取餐区等待，另一位服务员专门负责配餐。这体现了服务解耦，各自专注于特定环节，类似于生产者-消费者模式。

生产者-消费者模式该模式涉及两种角色：生产者和消费者。它们通过一个共享的内存缓冲区进行通信。生产者负责生产数据（或任务），并将数据放入缓冲区；消费者则从缓冲区取出数据进行处理。这种模式有效解耦了任务的产生和处理过程。

在实际应用中，例如用户注册服务，可以将账号验证、邮箱验证、短信验证等分解为独立的消费者服务。当用户提交注册信息后，主服务（生产者）将数据分发给这些消费者。如果消费者处理能力有限，可以引入请求队列（即内存缓冲区），由消息队列框架来管理。

模式实现生产者和消费者通常由多线程扮演，共享的内存缓冲区可以使用各种集合类实现。关键在于如何处理多线程间的同步与协作，尤其要解决以下问题：

当缓冲区为空时，消费者必须等待。
当缓冲区已满时，生产者必须等待。
在非满非空状态下，两者可动态平衡。
对共享缓冲区的读写操作必须是线程安全的，需要引入锁机制。

实现一：使用 BlockingQueue Java提供的 java.util.concurrent.BlockingQueue 接口及其实现（如 LinkedBlockingDeque）是实现生产者-消费者模式的理想选择。它内置了阻塞和唤醒机制，简化了多线程同步的复杂性。

生产者代码示例

package ProducerConsumer.Version1;

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 生产者线程
 */
public class DataProducer implements Runnable {
    private volatile boolean isActive = true;
    private final BlockingQueue<SharedData> buffer;
    private static final AtomicInteger producedCount = new AtomicInteger(0);
    private static final int PRODUCING_INTERVAL_MS = 1000;

    public DataProducer(BlockingQueue<SharedData> buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        SharedData dataItem;
        Random random = new Random();
        System.out.println("Producer started: " + Thread.currentThread().getId());
        try {
            while (isActive) {
                // 模拟生产耗时
                Thread.sleep(random.nextInt(PRODUCING_INTERVAL_MS));
                dataItem = new SharedData(producedCount.incrementAndGet());
                System.out.println("Produced: " + dataItem);
                // 尝试将数据放入缓冲区，若缓冲区满则等待
                if (!buffer.offer(dataItem, 2, TimeUnit.SECONDS)) {
                    System.err.println("Failed to add to buffer, buffer might be full.");
                }
            }
        } catch (InterruptedException e) {
            System.err.println("Producer interrupted.");
            Thread.currentThread().interrupt(); // Preserve interrupt status
        }
    }

    public void terminate() {
        isActive = false;
    }
}

消费者代码示例

package ProducerConsumer.Version1;

import java.text.MessageFormat;
import java.util.Random;
import java.util.concurrent.BlockingQueue;

/**
 * 消费者线程
 */
public class DataConsumer implements Runnable {
    private final BlockingQueue<SharedData> buffer;
    private static final int CONSUMING_INTERVAL_MS = 1000;

    public DataConsumer(BlockingQueue<SharedData> buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        System.out.println("Consumer started: " + Thread.currentThread().getId());
        Random random = new Random();
        try {
            while (true) {
                // take() 方法会阻塞直到缓冲区中有元素可用
                SharedData dataItem = buffer.take();
                if (dataItem != null) {
                    int value = dataItem.getValue();
                    int result = value * value; // 模拟处理：计算平方
                    System.out.println(MessageFormat.format("Consumed: {0}*{0}={1}", value, result));
                    // 模拟处理耗时
                    Thread.sleep(random.nextInt(CONSUMING_INTERVAL_MS));
                }
            }
        } catch (InterruptedException e) {
            System.err.println("Consumer interrupted.");
            Thread.currentThread().interrupt(); // Preserve interrupt status
        }
    }
}

主程序代码示例

package ProducerConsumer.Version1;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 主启动类
 */
public class MainApplication {
    public static void main(String[] args) throws InterruptedException {
        // 创建一个容量为10的阻塞队列
        BlockingQueue<SharedData> sharedBuffer = new LinkedBlockingDeque<>(10);

        // 创建生产者和消费者实例
        DataProducer producer1 = new DataProducer(sharedBuffer);
        DataProducer producer2 = new DataProducer(sharedBuffer);
        DataProducer producer3 = new DataProducer(sharedBuffer);

        DataConsumer consumer1 = new DataConsumer(sharedBuffer);
        DataConsumer consumer2 = new DataConsumer(sharedBuffer);
        DataConsumer consumer3 = new DataConsumer(sharedBuffer);

        // 使用线程池管理线程
        ExecutorService threadPool = Executors.newCachedThreadPool();

        // 提交任务
        threadPool.execute(producer1);
        threadPool.execute(producer2);
        threadPool.execute(producer3);
        threadPool.execute(consumer1);
        threadPool.execute(consumer2);
        threadPool.execute(consumer3);

        // 运行10秒后停止生产者
        Thread.sleep(10 * 1000);
        producer1.terminate();
        producer2.terminate();
        producer3.terminate();

        // 再等待3秒确保消费者处理完剩余数据
        Thread.sleep(3000);
        threadPool.shutdown(); // 关闭线程池
    }
}

数据载体类

package ProducerConsumer.Version1;

/**
 * 共享数据对象
 */
public class SharedData {
    private final int intValue;

    public SharedData(int value) {
        this.intValue = value;
    }

    public int getValue() {
        return intValue;
    }

    @Override
    public String toString() {
        return "Data(" + intValue + ")";
    }
}

BlockingQueue 的 put() 和 take() 方法会自动处理线程的阻塞和唤醒。当队列满时 put() 会阻塞，当队列空时 take() 会阻塞。这极大地简化了线程同步的实现。

实现二：使用 wait() 和 notifyAll() 在没有 BlockingQueue 的时代，开发者常使用 Object.wait() 和 Object.notifyAll() 方法来实现线程间的通信和同步。

生产者代码示例

package ProducerConsumer.Version2;

import java.util.List;
import java.util.Random;

/**
 * 生产者线程 (使用 wait/notifyAll)
 */
public class DataProducerWait implements Runnable {
    private final List<SharedDataItem> buffer;
    private final int capacity;

    public DataProducerWait(List<SharedDataItem> buffer, int capacity) {
        this.buffer = buffer;
        this.capacity = capacity;
    }

    @Override
    public void run() {
        try {
            while (true) {
                if (Thread.currentThread().isInterrupted()) break;
                Random random = new Random();
                long producedValue = random.nextInt(100);
                SharedDataItem newItem = new SharedDataItem(producedValue);

                synchronized (buffer) {
                    // 如果缓冲区已满，等待
                    if (buffer.size() >= capacity) {
                        System.out.println("Buffer full. Producer " + Thread.currentThread().getId() + " waiting.");
                        buffer.wait(); // 释放锁并等待
                    }
                    buffer.add(newItem);
                    System.out.println("Produced: " + producedValue + " by " + Thread.currentThread().getId());
                    buffer.notifyAll(); // 唤醒可能在等待的消费者
                }
                Thread.sleep(1000); // 模拟生产间隔
            }
        } catch (InterruptedException e) {
            System.err.println("Producer interrupted.");
            Thread.currentThread().interrupt();
        }
    }
}

消费者代码示例

package ProducerConsumer.Version2;

import java.util.List;

/**
 * 消费者线程 (使用 wait/notifyAll)
 */
public class DataConsumerWait implements Runnable {
    private final List<SharedDataItem> buffer;

    public DataConsumerWait(List<SharedDataItem> buffer) {
        this.buffer = buffer;
    }

    @Override
    public void run() {
        try {
            while (true) {
                if (Thread.currentThread().isInterrupted()) break;
                SharedDataItem consumedItem = null;
                synchronized (buffer) {
                    // 如果缓冲区为空，等待
                    if (buffer.isEmpty()) {
                        System.out.println("Buffer empty. Consumer " + Thread.currentThread().getId() + " waiting.");
                        buffer.wait(); // 释放锁并等待
                    }
                    consumedItem = buffer.remove(0); // 取出第一个元素
                    System.out.println("Consumed: " + consumedItem.getValue() + " by " + Thread.currentThread().getId());
                    buffer.notifyAll(); // 唤醒可能在等待的生产者
                }
                // 模拟处理时间
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            System.err.println("Consumer interrupted.");
            Thread.currentThread().interrupt();
        }
    }
}

主程序代码示例

package ProducerConsumer.Version2;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 主启动类 (使用 wait/notifyAll)
 */
public class MainApplicationWait {
    public static void main(String[] args) {
        List<SharedDataItem> sharedBuffer = new ArrayList<>();
        int bufferCapacity = 10;

        // 创建生产者和消费者实例
        DataProducerWait producer1 = new DataProducerWait(sharedBuffer, bufferCapacity);
        DataProducerWait producer2 = new DataProducerWait(sharedBuffer, bufferCapacity);
        DataConsumerWait consumer1 = new DataConsumerWait(sharedBuffer);
        DataConsumerWait consumer2 = new DataConsumerWait(sharedBuffer);

        ExecutorService threadPool = Executors.newCachedThreadPool();

        threadPool.execute(producer1);
        threadPool.execute(producer2);
        threadPool.execute(consumer1);
        threadPool.execute(consumer2);

        // 示例：可以添加逻辑来优雅地停止线程
    }
}

数据载体类

package ProducerConsumer.Version2;

/**
 * 共享数据项
 */
public class SharedDataItem {
    private long value;

    public SharedDataItem(long value) {
        this.value = value;
    }

    public long getValue() {
        return value;
    }
}

在此实现中，生产者和消费者需要显式地使用 synchronized 块来保护对共享 buffer 的访问，并在缓冲区状态不满足条件时调用 wait() 进入等待状态，通过 notifyAll() 唤醒其他线程。

实现三：使用 ReentrantLock 和 Condition Java 5 引入的 java.util.concurrent.locks 包提供了更灵活的锁机制，ReentrantLock 及其 Condition 对象可以替代 synchronized, wait(), notifyAll()。

生产者代码示例

package ProducerConsumer.Version3;

import java.util.List;
import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * 生产者线程 (使用 ReentrantLock/Condition)
 */
public class DataProducerLock implements Runnable {
    private final List<SharedDataItemLock> buffer;
    private final int capacity;
    private final Lock accessLock;
    private final Condition notFullCondition;

    public DataProducerLock(List<SharedDataItemLock> buffer, int capacity, Lock lock, Condition condition) {
        this.buffer = buffer;
        this.capacity = capacity;
        this.accessLock = lock;
        this.notFullCondition = condition;
    }

    @Override
    public void run() {
        try {
            while (true) {
                if (Thread.currentThread().isInterrupted()) break;
                Random random = new Random();
                SharedDataItemLock newItem = new SharedDataItemLock(random.nextInt(500));

                accessLock.lock(); // 获取锁
                try {
                    // 如果缓冲区已满，等待
                    if (buffer.size() >= capacity) {
                        System.out.println("Buffer full. Producer " + Thread.currentThread().getId() + " waiting.");
                        notFullCondition.await(); // 等待，并释放锁
                    }
                    buffer.add(newItem);
                    System.out.println("Produced: " + newItem.getValue() + " by " + Thread.currentThread().getId());
                    // 唤醒所有等待的消费者（因为可能有一个元素被添加）
                    notFullCondition.signalAll(); // 也可以使用 signal()
                } finally {
                    accessLock.unlock(); // 释放锁
                }
                Thread.sleep(1000); // 模拟生产间隔
            }
        } catch (InterruptedException e) {
            System.err.println("Producer interrupted.");
            Thread.currentThread().interrupt();
        }
    }
}

消费者代码示例

package ProducerConsumer.Version3;

import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * 消费者线程 (使用 ReentrantLock/Condition)
 */
public class DataConsumerLock implements Runnable {
    private final List<SharedDataItemLock> buffer;
    private final Lock accessLock;
    private final Condition notEmptyCondition;

    public DataConsumerLock(List<SharedDataItemLock> buffer, Lock lock, Condition condition) {
        this.buffer = buffer;
        this.accessLock = lock;
        this.notEmptyCondition = condition;
    }

    @Override
    public void run() {
        try {
            while (true) {
                if (Thread.currentThread().isInterrupted()) break;
                SharedDataItemLock consumedItem = null;

                accessLock.lock(); // 获取锁
                try {
                    // 如果缓冲区为空，等待
                    if (buffer.isEmpty()) {
                        System.out.println("Buffer empty. Consumer " + Thread.currentThread().getId() + " waiting.");
                        notEmptyCondition.await(); // 等待，并释放锁
                    }
                    consumedItem = buffer.remove(0);
                    System.out.println("Consumed: " + consumedItem.getValue() + " by " + Thread.currentThread().getId());
                    // 唤醒所有等待的生产者（因为可能有一个槽位被释放）
                    notEmptyCondition.signalAll(); // 也可以使用 signal()
                } finally {
                    accessLock.unlock(); // 释放锁
                }
                // 模拟处理时间
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            System.err.println("Consumer interrupted.");
            Thread.currentThread().interrupt();
        }
    }
}

主程序代码示例

package ProducerConsumer.Version3;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 主启动类 (使用 ReentrantLock/Condition)
 */
public class MainApplicationLock {
    public static void main(String[] args) {
        List<SharedDataItemLock> sharedBuffer = new ArrayList<>();
        int bufferCapacity = 10;

        // 创建锁和条件变量
        ReentrantLock bufferLock = new ReentrantLock();
        Condition bufferNotFull = bufferLock.newCondition(); // 生产者等待条件
        Condition bufferNotEmpty = bufferLock.newCondition(); // 消费者等待条件

        // 创建生产者和消费者实例
        DataProducerLock producer1 = new DataProducerLock(sharedBuffer, bufferCapacity, bufferLock, bufferNotFull);
        DataProducerLock producer2 = new DataProducerLock(sharedBuffer, bufferCapacity, bufferLock, bufferNotFull);
        DataConsumerLock consumer1 = new DataConsumerLock(sharedBuffer, bufferLock, bufferNotEmpty);
        DataConsumerLock consumer2 = new DataConsumerLock(sharedBuffer, bufferLock, bufferNotEmpty);

        ExecutorService threadPool = Executors.newCachedThreadPool();

        threadPool.execute(producer1);
        threadPool.execute(producer2);
        threadPool.execute(consumer1);
        threadPool.execute(consumer2);

        // 示例：可以添加逻辑来优雅地停止线程
    }
}

数据载体类

package ProducerConsumer.Version3;

/**
 * 共享数据项 (使用 ReentrantLock/Condition)
 */
public class SharedDataItemLock {
    private int data;

    public SharedDataItemLock(int data) {
        this.data = data;
    }

    public int getData() {
        return data;
    }
}

这种方式提供了更细粒度的控制，可以为不同的等待条件创建不同的 Condition 对象，提高了并发性能和灵活性。

流水线式并行计算生产者-消费者模式的思想也广泛应用于流水线（Pipeline）并行计算。将一个复杂的计算任务分解为一系列串行的子任务，每个子任务由一个独立的线程（或进程）负责，并使用缓冲区（流水线）进行中间结果的传递。这种方法适用于满足"分治"原则且各阶段之间相互独立的计算。例如，计算 (A + B) * C：

P1 (生产者): 计算 D = A + B
P2 (消费者): 计算 R = D * C P1 生产数据（D），P2 消费数据（D）并进行计算。P1 是 P2 的生产者，而 P2 则是最终结果的生产者。流水线本身即为内存缓冲区。

总结：学习技术模型时，将其与现实生活中的类比相结合，能够加深理解。生产者-消费者模式不仅在并发编程中有重要应用，其解耦和异步处理的思想也适用于管理学等其他领域。

标签: Java 多线程并发

返回列表

上一篇：基于大语言模型的自主智能体架构与核心机制解析

下一篇：Apache Gravitino 元数据湖部署与配置实战

老程序员博客

Java生产者-消费者模式实现详解

相关文章

Linux crontab 详解

富文本里可以允许的 HTML 属性

Mac 安装 Node.js 指南

Dom\HTML_NO_DEFAULT_NS 的副作用：自动加闭合标签

Laravel 事件和监听器创建

自定义域名解析神器 dnsmasq

发表评论

Copyright © agingcoder.cn

Powered By Z-BlogPHP. Theme by TOYEAN.

老程序员博客

Java生产者-消费者模式实现详解

相关文章

Linux crontab 详解

富文本里可以允许的 HTML 属性

Mac 安装 Node.js 指南

Dom\HTML_NO_DEFAULT_NS 的副作用：自动加闭合标签

Laravel 事件和监听器创建

自定义域名解析神器 dnsmasq

发表评论取消回复

Copyright © agingcoder.cn

Powered By Z-BlogPHP. Theme by TOYEAN.

发表评论