Java实现生产者和消费者
前言
生产者和消费者问题是线程模型中的经典问题：生产者和消费者在同一时间段内共用同一个存储空间，生产者往存储空间中添加产品，消费者从存储空间中取走产品，当存储空间为空时，消费者阻塞，当存储空间满时，生产者阻塞。
阻塞队列
1.LinkedBlockingQueue
LinkedBlockingQueue是线程安全的、有界队列（可以指定容量，不指定时默认最大是Integer.MAX_VALUE）。
核心思路是在入队、出队的时候通过锁来保证线程安全。主要用到put和take方法，put方法在队列满的时候会阻塞直到有队列成员被消费，take方法在队列空的时候会阻塞，直到有队列成员被放进来。
package com.haodf.master.thread.threadqueue;import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
/**
* @author Huan Lee
* @version 1.0
* @date 2020-07-10 14:22
* @describtion 业精于勤，荒于嬉；行成于思，毁于随。
*/
public class ThreadQueueTest {
/**
* 定义装苹果的篮子
*/
public class Basket {
// 篮子，能够容纳2个苹果
BlockingQueue<String> basket = new LinkedBlockingQueue<>(2);
// 生产苹果，放入篮子
public void produce() throws InterruptedException {
// put方法放入一个苹果，若basket满了，等到basket有位置
basket.put("An apple");
}
// 消费苹果，从篮子中取走
public String consume() throws InterruptedException {
// take方法取出一个苹果，若basket为空，等到basket有苹果为止(获取并移除此队列的头部)
System.out.println("队列大小："+basket.size());
return basket.take();
}
}
/**
* 定义苹果生产者
*/
class Producer implements Runnable {
private String instance;
private Basket basket;
public Producer(String instance, Basket basket) {
this.instance = instance;
this.basket = basket;
}
@Override
public void run() {
try {
while (true) {
// 生产苹果
basket.produce();
System.out.println("生产者生产苹果完毕：" + instance);
// 休眠3000ms
Thread.sleep(3000);
}
} catch (InterruptedException ex) {
System.out.println("Producer Interrupted");
}
}
}
/**
* 定义苹果消费者
*/
class Consumer implements Runnable {
private String instance;
private Basket basket;
public Consumer(String instance, Basket basket) {
this.instance = instance;
this.basket = basket;
}
@Override
public void run() {
try {
while (true) {
// 消费苹果
basket.consume();
System.out.println("消费者消费苹果完毕：" + instance);
System.out.println("***************");
// 休眠3000ms
Thread.sleep(3000);
}
} catch (InterruptedException ex) {
System.out.println("Consumer Interrupted");
}
}
}
public static void main(String[] args) {
ThreadQueueTest test = new ThreadQueueTest();
// 建立一个装苹果的篮子
Basket basket = test.new Basket();
ExecutorService service = Executors.newCachedThreadPool();
Producer producer = test.new Producer("生产者001", basket);
Producer producer2 = test.new Producer("生产者002", basket);
Consumer consumer = test.new Consumer("消费者001", basket);
service.submit(producer);
service.submit(producer2);
service.submit(consumer);
}
}
输出：
生产者生产苹果完毕：生产者002
队列大小：2
生产者生产苹果完毕：生产者001
消费者消费苹果完毕：消费者001
***************
队列大小：1
生产者生产苹果完毕：生产者002
消费者消费苹果完毕：消费者001
***************
生产者生产苹果完毕：生产者001
队列大小：2
消费者消费苹果完毕：消费者001
***************
2.ArrayBlockingQueue
BlockingQueue即阻塞队列，从阻塞这个词可以看出，在某些情况下对阻塞队列的访问可能会造成阻塞。阻塞队列是线程安全的。被阻塞的情况主要有如下两种:
- 当队列满了的时候进行入队列操作
- 当队列空了的时候进行出队列操作
因此，当一个线程对已经满了的阻塞队列进行入队操作时会阻塞，除非有另外一个线程进行了出队操作，当一个线程对一个空的阻塞队列进行出队操作时也会阻塞，除非有另外一个线程进行了入队操作。
这四类方法分别对应的是：
1 . ThrowsException：如果操作不能马上进行，则抛出异常
2 . SpecialValue：如果操作不能马上进行，将会返回一个特殊的值，一般是true或者false
3 . Blocks:如果操作不能马上进行，操作会被阻塞
4 . TimesOut:如果操作不能马上进行，操作会被阻塞指定的时间，如果指定时间没执行，则返回一个特殊值，一般是true或者false
下面来看由阻塞队列实现的生产者消费者模型,这里我们使用take()和put()方法，这里生产者和生产者，消费者和消费者之间不存在同步，所以会出现连续生成和连续消费的现象
package com.haodf.master.thread.threadqueue;import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
* @author Huan Lee
* @version 1.0
* @date 2020-07-10 15:52
* @describtion 业精于勤，荒于嬉；行成于思，毁于随。
*/
public class ArrayBlockingQueueTest {
private static Integer count = 0;
/**
* 创建一个阻塞队列
*/
final BlockingQueue blockingQueue = new ArrayBlockingQueue(2);
public static void main(String[] args) {
ArrayBlockingQueueTest test3 = new ArrayBlockingQueueTest();
new Thread(test3.new Producer()).start();
new Thread(test3.new Producer()).start();
new Thread(test3.new Consumer()).start();
}
class Producer implements Runnable {
@Override
public void run() {
for (int i = 0; i < 5; i++) {
try {
Thread.sleep(3000);
} catch (Exception e) {
e.printStackTrace();
}
try {
blockingQueue.put(1);
count++;
System.out.println(Thread.currentThread().getName()
+ "生产者生产，目前总共有" + count);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
class Consumer implements Runnable {
@Override
public void run() {
for (int i = 0; i < 5; i++) {
try {
Thread.sleep(3000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
try {
blockingQueue.take();
count--;
System.out.println(Thread.currentThread().getName()
+ "消费者消费，目前总共有" + count);
System.out.println("***************");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
输出：
Thread-0生产者生产，目前总共有1
Thread-1生产者生产，目前总共有1
Thread-2消费者消费，目前总共有1
***************
Thread-0生产者生产，目前总共有2
Thread-2消费者消费，目前总共有1
***************
Thread-1生产者生产，目前总共有2
Thread-2消费者消费，目前总共有1
***************
Thread-1生产者生产，目前总共有2
Thread-2消费者消费，目前总共有1
***************
wait()和notify()实现
这也是最简单最基础的实现，缓冲区满和为空时都调用wait()方法等待，当生产者生产了一个产品或者消费者消费了一个产品之后会唤醒所有线程。
/** * 生产者和消费者，wait()和notify()的实现
*/
public class Test1 {
private static Integer count = 0;
private static final Integer FULL = 10;
private static String LOCK = "lock";
public static void main(String[] args) {
Test1 test1 = new Test1();
new Thread(test1.new Producer()).start();
new Thread(test1.new Consumer()).start();
new Thread(test1.new Producer()).start();
new Thread(test1.new Consumer()).start();
new Thread(test1.new Producer()).start();
new Thread(test1.new Consumer()).start();
new Thread(test1.new Producer()).start();
new Thread(test1.new Consumer()).start();
}
class Producer implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
try {
Thread.sleep(3000);
} catch (Exception e) {
e.printStackTrace();
}
synchronized (LOCK) {
while (count == FULL) {
try {
LOCK.wait();
} catch (Exception e) {
e.printStackTrace();
}
}
count++;
System.out.println(Thread.currentThread().getName() + "生产者生产，目前总共有" + count);
LOCK.notifyAll();
}
}
}
}
class Consumer implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized (LOCK) {
while (count == 0) {
try {
LOCK.wait();
} catch (Exception e) {
}
}
count--;
System.out.println(Thread.currentThread().getName() + "消费者消费，目前总共有" + count);
LOCK.notifyAll();
}
}
}
}
}
输出：
Thread-0生产者生产，目前总共有1
Thread-4生产者生产，目前总共有2
Thread-3消费者消费，目前总共有1
Thread-1消费者消费，目前总共有0
Thread-2生产者生产，目前总共有1
Thread-6生产者生产，目前总共有2
Thread-7消费者消费，目前总共有1
Thread-5消费者消费，目前总共有0
Thread-0生产者生产，目前总共有1
Thread-4生产者生产，目前总共有2
Thread-3消费者消费，目前总共有1
Thread-6生产者生产，目前总共有2
可重入锁ReentrantLock实现
java.util.concurrent.lock 中的 Lock 框架是锁定的一个抽象，通过对lock的lock()方法和unlock()方法实现了对锁的显示控制，而synchronize()则是对锁的隐性控制。
可重入锁，也叫做递归锁，指的是同一线程 外层函数获得锁之后 ，内层递归函数仍然有获取该锁的代码，但不受影响，简单来说，该锁维护这一个与获取锁相关的计数器，如果拥有锁的某个线程再次得到锁，那么获取计数器就加1，函数调用结束计数器就减1，然后锁需要被释放两次才能获得真正释放。已经获取锁的线程进入其他需要相同锁的同步代码块不会被阻塞。
import java.util.concurrent.locks.Condition;import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 生产者和消费者，ReentrantLock的实现
*/
public class Test2 {
private static Integer count = 0;
private static final Integer FULL = 10;
//创建一个锁对象
private Lock lock = new ReentrantLock();
//创建两个条件变量，一个为缓冲区非满，一个为缓冲区非空
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
public static void main(String[] args) {
Test2 test2 = new Test2();
new Thread(test2.new Producer()).start();
new Thread(test2.new Consumer()).start();
new Thread(test2.new Producer()).start();
new Thread(test2.new Consumer()).start();
new Thread(test2.new Producer()).start();
new Thread(test2.new Consumer()).start();
new Thread(test2.new Producer()).start();
new Thread(test2.new Consumer()).start();
}
class Producer implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
try {
Thread.sleep(3000);
} catch (Exception e) {
e.printStackTrace();
}
//获取锁
lock.lock();
try {
while (count == FULL) {
try {
notFull.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
count++;
System.out.println(Thread.currentThread().getName()
+ "生产者生产，目前总共有" + count);
//唤醒消费者
notEmpty.signal();
} finally {
//释放锁
lock.unlock();
}
}
}
}
class Consumer implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
try {
Thread.sleep(3000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
lock.lock();
try {
while (count == 0) {
try {
notEmpty.await();
} catch (Exception e) {
e.printStackTrace();
}
}
count--;
System.out.println(Thread.currentThread().getName()
+ "消费者消费，目前总共有" + count);
notFull.signal();
} finally {
lock.unlock();
}
}
}
}
}
输出：
信号量Semaphore实现
Semaphore（信号量）是用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源，在操作系统中是一个非常重要的问题，可以用来解决哲学家就餐问题。Java中的Semaphore维护了一个许可集，一开始先设定这个许可集的数量，可以使用acquire()方法获得一个许可，当许可不足时会被阻塞，release()添加一个许可。在下列代码中，还加入了另外一个mutex信号量，维护生产者消费者之间的同步关系，保证生产者和消费者之间的交替进行
import java.util.concurrent.Semaphore;/**
* 使用semaphore信号量实现
*/
public class Test4 {
private static Integer count = 0;
//创建三个信号量
final Semaphore notFull = new Semaphore(10);
final Semaphore notEmpty = new Semaphore(0);
final Semaphore mutex = new Semaphore(1);
public static void main(String[] args) {
Test4 test4 = new Test4();
new Thread(test4.new Producer()).start();
new Thread(test4.new Consumer()).start();
new Thread(test4.new Producer()).start();
new Thread(test4.new Consumer()).start();
new Thread(test4.new Producer()).start();
new Thread(test4.new Consumer()).start();
new Thread(test4.new Producer()).start();
new Thread(test4.new Consumer()).start();
}
class Producer implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
try {
notFull.acquire();
mutex.acquire();
count++;
System.out.println(Thread.currentThread().getName()
+ "生产者生产，目前总共有" + count);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
mutex.release();
notEmpty.release();
}
}
}
}
class Consumer implements Runnable {
@Override
public void run() {
for (int i = 0; i < 10; i++) {
try {
Thread.sleep(3000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
try {
notEmpty.acquire();
mutex.acquire();
count--;
System.out.println(Thread.currentThread().getName()
+ "消费者消费，目前总共有" + count);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
mutex.release();
notFull.release();
}
}
}
}
}
输出：
管道输入输出流PipedInputStream和PipedOutputStream实现
在java的io包下，PipedOutputStream和PipedInputStream分别是管道输出流和管道输入流。
它们的作用是让多线程可以通过管道进行线程间的通讯。在使用管道通信时，必须将PipedOutputStream和PipedInputStream配套使用。
使用方法：先创建一个管道输入流和管道输出流，然后将输入流和输出流进行连接，用生产者线程往管道输出流中写入数据，消费者在管道输入流中读取数据，这样就可以实现了不同线程间的相互通讯，但是这种方式在生产者和生产者、消费者和消费者之间不能保证同步，也就是说在一个生产者和一个消费者的情况下是可以生产者和消费者之间交替运行的，多个生成者和多个消费者者之间则不行
/** * 使用管道实现生产者消费者模型
*/
public class Test5 {
final PipedInputStream pis = new PipedInputStream();
final PipedOutputStream pos = new PipedOutputStream();
{
try {
pis.connect(pos);
} catch (IOException e) {
e.printStackTrace();
}
}
class Producer implements Runnable {
@Override
public void run() {
try {
while(true) {
Thread.sleep(1000);
int num = (int) (Math.random() * 255);
System.out.println(Thread.currentThread().getName() + "生产者生产了一个数字，该数字为： " + num);
pos.write(num);
pos.flush();
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
pos.close();
pis.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
class Consumer implements Runnable {
@Override
public void run() {
try {
while(true) {
Thread.sleep(1000);
int num = pis.read();
System.out.println("消费者消费了一个数字，该数字为：" + num);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
pos.close();
pis.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
public static void main(String[] args) {
Test5 test5 = new Test5();
new Thread(test5.new Producer()).start();
new Thread(test5.new Consumer()).start();
}
}
以上是 Java实现生产者和消费者 的全部内容， 来源链接： utcz.com/z/518199.html
