首页 > 编程语言 >【转】-Java实现生产者和消费者的5种方式

【转】-Java实现生产者和消费者的5种方式

时间：2024-07-10 09:54:06浏览次数：13  
标签：count 总共 Java 消费者 Thread 生产者 start new

Java实现生产者和消费者的5种方式

该博客转载自​掘金​的​Java实现生产者和消费者的5种方式

1. 前言

生产者和消费者问题是线程模型中的经典问题：生产者和消费者在同一时间段内共用同一个存储空间，生产者往存储空间中添加产品，消费者从存储空间中取走产品，当存储空间为空时，消费者阻塞，当存储空间满时，生产者阻塞。

1706867654525.webp

现在用四种方式来实现生产者消费者模型

2. wait()和notify()方法的实现

这也是最简单最基础的实现，缓冲区满和为空时都调用wait()方法等待，当生产者生产了一个产品或者消费者消费了一个产品之后会唤醒所有线程。

/**
 * 生产者和消费者，wait()和notify()的实现
 * @author ZGJ
 * @date 2017年6月22日
 */
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
Thread-1消费者消费，目前总共有1
Thread-7消费者消费，目前总共有0
Thread-2生产者生产，目前总共有1
Thread-5消费者消费，目前总共有0
Thread-0生产者生产，目前总共有1
Thread-4生产者生产，目前总共有2
Thread-3消费者消费，目前总共有1
Thread-7消费者消费，目前总共有0
Thread-6生产者生产，目前总共有1
Thread-2生产者生产，目前总共有2
Thread-1消费者消费，目前总共有1
Thread-5消费者消费，目前总共有0
Thread-0生产者生产，目前总共有1
Thread-4生产者生产，目前总共有2
Thread-3消费者消费，目前总共有1
Thread-1消费者消费，目前总共有0
Thread-6生产者生产，目前总共有1
Thread-7消费者消费，目前总共有0
Thread-2生产者生产，目前总共有1

3. 可重入锁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的实现
 * 
 * @author ZGJ
 * @date 2017年6月22日
 */
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();
                }
            }
        }
    }
}

4. 阻塞队列BlockingQueue的实现

BlockingQueue即阻塞队列，从阻塞这个词可以看出，在某些情况下对阻塞队列的访问可能会造成阻塞。被阻塞的情况主要有如下两种:

  1. 当队列满了的时候进行入队列操作
  2. 当队列空了的时候进行出队列操作
    因此，当一个线程对已经满了的阻塞队列进行入队操作时会阻塞，除非有另外一个线程进行了出队操作，当一个线程对一个空的阻塞队列进行出队操作时也会阻塞，除非有另外一个线程进行了入队操作。
    从上可知，阻塞队列是线程安全的。
    下面是BlockingQueue接口的一些方法:
操作 抛异常 特定值 阻塞 超时
插入 add(o) offer(o) put(o) offer(o, timeout, timeunit)
移除 remove(o) poll(o) take(o) poll(timeout, timeunit)
检查 element(o) peek(o)

这四类方法分别对应的是：

  1. ThrowsException：如果操作不能马上进行，则抛出异常
  2. SpecialValue：如果操作不能马上进行，将会返回一个特殊的值，一般是true或者false
  3. Blocks:如果操作不能马上进行，操作会被阻塞
  4. TimesOut:如果操作不能马上进行，操作会被阻塞指定的时间，如果指定时间没执行，则返回一个特殊值，一般是true或者false

下面来看由阻塞队列实现的生产者消费者模型,这里我们使用take()和put()方法，这里生产者和生产者，消费者和消费者之间不存在同步，所以会出现连续生成和连续消费的现象

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
/**
 * 使用BlockingQueue实现生产者消费者模型
 * @author ZGJ
 * @date 2017年6月29日
 */
public class Test3 {
    private static Integer count = 0;
    //创建一个阻塞队列
    final BlockingQueue blockingQueue = new ArrayBlockingQueue<>(10);
    public static void main(String[] args) {
        Test3 test3 = new Test3();
        new Thread(test3.new Producer()).start();
        new Thread(test3.new Consumer()).start();
        new Thread(test3.new Producer()).start();
        new Thread(test3.new Consumer()).start();
        new Thread(test3.new Producer()).start();
        new Thread(test3.new Consumer()).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 < 10; 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 < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                try {
                    blockingQueue.take();
                    count--;
                    System.out.println(Thread.currentThread().getName()
                            + "消费者消费，目前总共有" + count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

5. 信号量Semaphore的实现

Semaphore（信号量）是用来控制同时访问特定资源的线程数量，它通过协调各个线程，以保证合理的使用公共资源，在操作系统中是一个非常重要的问题，可以用来解决哲学家就餐问题。Java中的Semaphore维护了一个许可集，一开始先设定这个许可集的数量，可以使用acquire()方法获得一个许可，当许可不足时会被阻塞，release()添加一个许可。在下列代码中，还加入了另外一个mutex信号量，维护生产者消费者之间的同步关系，保证生产者和消费者之间的交替进行

import java.util.concurrent.Semaphore;
/**
 * 使用semaphore信号量实现
 * @author ZGJ
 * @date 2017年6月29日
 */
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();
                }
            }
        }
    }
}

6. 管道输入输出流PipedInputStream和PipedOutputStream实现

在java的io包下，PipedOutputStream和PipedInputStream分别是管道输出流和管道输入流。
它们的作用是让多线程可以通过管道进行线程间的通讯。在使用管道通信时，必须将PipedOutputStream和PipedInputStream配套使用。

使用方法：先创建一个管道输入流和管道输出流，然后将输入流和输出流进行连接，用生产者线程往管道输出流中写入数据，消费者在管道输入流中读取数据，这样就可以实现了不同线程间的相互通讯，但是这种方式在生产者和生产者、消费者和消费者之间不能保证同步，也就是说在一个生产者和一个消费者的情况下是可以生产者和消费者之间交替运行的，多个生成者和多个消费者者之间则不行

/**
 * 使用管道实现生产者消费者模型
 * @author ZGJ
 * @date 2017年6月30日
 */
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();
    }
}

标签：count,总共,Java,消费者,Thread,生产者,start,new
From： https://www.cnblogs.com/booleandev/p/18293254/turn-the-5-ways-of-realizing-producers-and-c

相关文章

  • Java中的线程池详解
    Java中的线程池详解大家好，我是微赚淘客系统3.0的小编，是个冬天不穿秋裤，天冷也要风度的程序猿！1.理解线程池的概念在Java中，线程池是一种管理和复用线程的机制，它能够有效地控制并发线程数量，提高系统性能和资源利用率。使用线程池可以避免反复创建和销毁线程的开销，同时能够管理大量......
  • 【转】-Java锁解析
    不可不说的Java“锁”事**该博客转载自美团技术团队的不可不说的Java“锁”事**前言Java提供了种类丰富的锁，每种锁因其特性的不同，在适当的场景下能够展现出非常高的效率。本文旨在对锁相关源码（本文中的源码来自JDK8和Netty3.10.6）、使用场景进行举例，为读者介绍主流锁的知识......
  • 【转】-java多线程读写锁ReadWriteLock
    Java多线程中读写锁ReadWriteLock的使用该博客转载自​lavimer​的​Java多线程中读写锁ReadWriteLock的使用1.概念读写锁分为读锁和写锁，多个读锁之间是不需要互斥的(读操作不会改变数据，如果上了锁，反而会影响效率)，写锁和写锁之间需要互斥，也就是说，如果只是读数据，就可以多个线程......
  • [Java SE] 字节操作工具类：ByteUtils
    0引言与嵌入式软件数据交互过程中，必然涉及各种的、大量的字节操作场景。如：16进制与10进制、2进制间的转换，字符串、byte数组与int之间的转换等。故此有此核心工具类的沉淀。1ByteUtils依赖<properties> <!--编程提效工具--> <lombok.version>1.18.22</lombok.version>......
  • 【转】-Java并发编程：CountDownLatch、CyclicBarrier和Semaphore
    Java并发编程：CountDownLatch、CyclicBarrier和Semaphore该博客转载自​Matrix海子​的​Java并发编程：CountDownLatch、CyclicBarrier和Semaphore在java1.5中，提供了一些非常有用的辅助类来帮助我们进行并发编程，比如CountDownLatch，CyclicBarrier和Semaphore，今天我们就来学习一下......
  • 【转】-Java并发之CyclicBarrier
    Java并发之CyclicBarrier​ 该博客转载自​巴蜀码哥​**的​Java并发之CyclicBarrier**barrier(屏障)与互斥量、读写锁、自旋锁不同，它不是用来保护临界区的。相反，它跟条件变量一样，是用来协同多线程一起工作的。条件变量是多线程间传递状态的改变来达到协同工作的效果。屏障是......
  • 深入理解Java中的并发编程
    深入理解Java中的并发编程大家好，我是微赚淘客系统3.0的小编，是个冬天不穿秋裤，天冷也要风度的程序猿！并发编程是Java开发中的一个重要领域，通过并发编程，可以提高程序的执行效率和资源利用率。本文将深入探讨Java中的并发编程，包括线程的创建、同步机制、并发集合、线程池和并发工具类......
  • Java中的垃圾回收机制详解
    Java中的垃圾回收机制详解大家好，我是微赚淘客系统3.0的小编，是个冬天不穿秋裤，天冷也要风度的程序猿！Java中的垃圾回收（GarbageCollection,GC）机制是Java虚拟机（JVM）管理内存的一项重要功能。GC机制通过自动回收不再使用的对象所占用的内存，防止内存泄漏，提升应用程序的性能和稳定性。......
  • 使用JUnit进行Java单元测试
    使用JUnit进行Java单元测试大家好，我是微赚淘客系统3.0的小编，是个冬天不穿秋裤，天冷也要风度的程序猿！单元测试是软件开发过程中至关重要的一部分，能够确保代码的正确性、健壮性和可维护性。JUnit是一个流行的Java单元测试框架，广泛应用于Java应用程序的测试。本文将详细介绍如......
  • 如何在Java中处理JSON数据
    如何在Java中处理JSON数据大家好，我是微赚淘客系统3.0的小编，是个冬天不穿秋裤，天冷也要风度的程序猿！在现代Web开发中，JSON（JavaScriptObjectNotation）已经成为数据交换的标准格式之一。Java提供了多种库来处理JSON数据，最常用的包括Gson、Jackson和org.json等。在本文中，我们将详细介......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99