java 生产者消费者问题(Java生产者消费者模式详解与应用)

原创

ithorizon 6个月前 (10-21) 阅读数 15 #后端开发

Java 生产者消费者问题详解与应用

一、生产者消费者问题概述

生产者消费者问题是多线程编程中的一个经典问题，核心涉及到线程之间的同步和通信。问题描述如下：有一个生产者线程和多个消费者线程，生产者负责生产数据，消费者负责消费这些数据。为了保证数据的一致性和正确性，需要通过某种机制来保证生产者和消费者之间的同步。

二、解决方案

在Java中，解决生产者消费者问题核心有以下几种方案：

使用synchronized关键字和wait/notify/notifyAll方法

使用ReentrantLock和Condition

使用BlockingQueue

三、使用synchronized关键字和wait/notify/notifyAll方法

下面通过一个易懂的示例来讲解使用synchronized关键字和wait/notify/notifyAll方法解决生产者消费者问题。

3.1 数据缓冲区类

数据缓冲区类用于存储生产者生产的数据，同时提供方法供消费者消费数据。


public class DataBuffer {
    private int[] buffer = new int[10];
    private int count = 0;
    private int putIndex = 0;
    private int takeIndex = 0;
    public synchronized void put(int value) throws InterruptedException {
        while (count == buffer.length) {
            this.wait();
        }
        buffer[putIndex] = value;
        putIndex = (putIndex + 1) % buffer.length;
        count++;
        this.notifyAll();
    }
    public synchronized int take() throws InterruptedException {
        while (count == 0) {
            this.wait();
        }
        int value = buffer[takeIndex];
        takeIndex = (takeIndex + 1) % buffer.length;
        count--;
        this.notifyAll();
        return value;
    }
}

3.2 生产者类

生产者类负责生产数据，并将数据放入数据缓冲区。


public class Producer implements Runnable {
    private DataBuffer buffer;
    public Producer(DataBuffer buffer) {
        this.buffer = buffer;
    }
    @Override
    public void run() {
        for (int i = 0; i < 100; i++) {
            try {
                buffer.put(i);
                System.out.println("生产者生产数据：" + i);
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

3.3 消费者类

消费者类负责从数据缓冲区中消费数据。


public class Consumer implements Runnable {
    private DataBuffer buffer;
    public Consumer(DataBuffer buffer) {
        this.buffer = buffer;
    }
    @Override
    public void run() {
        while (true) {
            try {
                int data = buffer.take();
                System.out.println("消费者消费数据：" + data);
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

3.4 主程序

主程序负责创建线程并启动线程。


public class Main {
    public static void main(String[] args) {
        DataBuffer buffer = new DataBuffer();
        Thread producerThread = new Thread(new Producer(buffer));
        Thread consumerThread = new Thread(new Consumer(buffer));
        producerThread.start();
        consumerThread.start();
    }
}

四、使用ReentrantLock和Condition

除了使用synchronized关键字和wait/notify/notifyAll方法，我们还可以使用ReentrantLock和Condition来解决生产者消费者问题。

4.1 数据缓冲区类

使用ReentrantLock和Condition重写数据缓冲区类。


import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class DataBuffer {
    private int[] buffer = new int[10];
    private int count = 0;
    private int putIndex = 0;
    private int takeIndex = 0;
    private ReentrantLock lock = new ReentrantLock();
    private Condition notFull = lock.newCondition();
    private Condition notEmpty = lock.newCondition();
    public void put(int value) throws InterruptedException {
        lock.lock();
        try {
            while (count == buffer.length) {
                notFull.await();
            }
            buffer[putIndex] = value;
            putIndex = (putIndex + 1) % buffer.length;
            count++;
            notEmpty.signalAll();
        } finally {
            lock.unlock();
        }
    }
    public int take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {
                notEmpty.await();
            }
            int value = buffer[takeIndex];
            takeIndex = (takeIndex + 1) % buffer.length;
            count--;
            notFull.signalAll();
            return value;
        } finally {
            lock.unlock();
        }
    }
}

五、使用BlockingQueue

Java提供了BlockingQueue接口，该接口的实现类已经实现了生产者消费者模式。常用的实现类有ArrayBlockingQueue和LinkedBlockingQueue。

5.1 数据缓冲区类

使用ArrayBlockingQueue作为数据缓冲区。


import java.util.concurrent.ArrayBlockingQueue;
public class DataBuffer {
    private ArrayBlockingQueue queue = new ArrayBlockingQueue<>(10);
    public void put(int value) throws InterruptedException {
        queue.put(value);
    }
    public int take() throws InterruptedException {
        return queue.take();
    }
}