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

更新时间：2024年07月04日 10:39:34 作者：xindoo

这篇文章主要介绍了Java生产者消费者的三种实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教

Java生产者消费者的实现

Java生产者消费者是最基础的线程同步问题，java岗面试中还是很容易遇到的，之前没写过多线程的代码，面试中被问到很尬啊，面完回来恶补下。

在网上查到大概有5种生产者消费者的写法，分别如下。

用synchronized对存储加锁，然后用object原生的wait() 和 notify()做同步。
用concurrent.locks.Lock，然后用condition的await() 和signal()做同步。
直接使用concurrent.BlockingQueue。
使用PipedInputStream/PipedOutputStream。
使用信号量semaphore。

我的理解，生产者消费者模式，其实只要保证在存储端同一时刻只有一个线程读或写就不会有问题，然后再去考虑线程同步。

方法1 2 5都比较类似，都是加锁来限制同一时刻只能有一个读或写。而方法3 4其实是在存储内部去保证读和写的唯一的，最低层肯定还是通过锁机制来实现的，java底层代码都封装好了而已。　　

我自己尝试写了下前三种，代码如下：　

synchronized版本

import java.util.LinkedList;
import java.util.Queue;

public class ProducerAndConsumer {
    private final int MAX_LEN = 10;
    private Queue<Integer> queue = new LinkedList<Integer>();
    class Producer extends Thread {
        @Override
        public void run() {
            producer();
        }
        private void producer() {
            while(true) {
                synchronized (queue) {
                    while (queue.size() == MAX_LEN) {
                        queue.notify();
                        System.out.println("当前队列满");
                        try {
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    queue.add(1);
                    queue.notify();
                    System.out.println("生产者生产一条任务，当前队列长度为" + queue.size());
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
    class Consumer extends Thread {
        @Override
        public void run() {
            consumer();
        }
        private void consumer() {
            while (true) {
                synchronized (queue) {
                    while (queue.size() == 0) {
                        queue.notify();
                        System.out.println("当前队列为空");
                        try {
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    queue.poll();
                    queue.notify();
                    System.out.println("消费者消费一条任务，当前队列长度为" + queue.size());
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
    public static void main(String[] args) {
        ProducerAndConsumer pc = new ProducerAndConsumer();
        Producer producer = pc.new Producer();
        Consumer consumer = pc.new Consumer();
        producer.start();
        consumer.start();
    }
}

lock版实现，使用了condition做线程之间的同步

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * version 1 doesn't use synchronized to improve performance
 */
public class ProducerAndConsumer1 {
    private final int MAX_LEN = 10;
    private Queue<Integer> queue = new LinkedList<Integer>();
    private final Lock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();
    class Producer extends Thread {
        @Override
        public void run() {
            producer();
        }
        private void producer() {
            while(true) {
                lock.lock();
                try {
                    while (queue.size() == MAX_LEN) {
                        System.out.println("当前队列满");
                        try {
                            condition.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    queue.add(1);
                    condition.signal();
                    System.out.println("生产者生产一条任务，当前队列长度为" + queue.size());
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } finally {
                    lock.unlock();
                }
            }
        }
    }
    class Consumer extends Thread {
        @Override
        public void run() {
            consumer();
        }
        private void consumer() {
            while (true) {
                lock.lock();
                try {
                    while (queue.size() == 0) {
                        System.out.println("当前队列为空");
                        try {
                            condition.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    queue.poll();
                    condition.signal();
                    System.out.println("消费者消费一条任务，当前队列长度为" + queue.size());
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } finally {
                    lock.unlock();
                }
            }
        }
    }
    public static void main(String[] args) {
        ProducerAndConsumer pc = new ProducerAndConsumer();
        Producer producer = pc.new Producer();
        Consumer consumer = pc.new Consumer();
        producer.start();
        consumer.start();
    }
}

BlockingQueue版实现

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;


public class ProducerAndConsumer {
    private BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>(10);
    class Producer extends Thread {
        @Override
        public void run() {
            producer();
        }
        private void producer() {
            while(true) {
                try {
                    queue.put(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("生产者生产一条任务，当前队列长度为" + queue.size());
                try {
                    Thread.sleep(new Random().nextInt(1000)+500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    class Consumer extends Thread {
        @Override
        public void run() {
            consumer();
        }
        private void consumer() {
            while (true) {
                try {
                    queue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者消费一条任务，当前队列长度为" + queue.size());
                try {
                    Thread.sleep(new Random().nextInt(1000)+500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    public static void main(String[] args) {
        ProducerAndConsumer pc = new ProducerAndConsumer();
        Producer producer = pc.new Producer();
        Consumer consumer = pc.new Consumer();
        producer.start();
        consumer.start();
    }
}

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

SpringBoot实现隐式参数注入的完整指南
这篇文章主要为大家详细介绍了SpringBoot实现隐式参数注入的相关方法,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下
2025-11-11
Mybatis批量插入返回插入成功后的主键id操作
这篇文章主要介绍了Mybatis批量插入返回插入成功后的主键id操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2021-03-03
Java异常处理方法汇总
这篇文章主要介绍了Java异常处理方法汇总，我们在软件开发的过程中，任何语言的开发过程中都离不开异常处理。下面下小编加来给大家分享各种异常处理，希望对大家有所帮助,需要的朋友可以参考一下
2021-12-12
使用Javacsv.jar的jar包操作csv文件的方法
使用javacsv的jar包很方便的可以操作csv文件。下面通过本教程给大家介绍使用javacsv.jar操作csv文件的方法，感兴趣的朋友一起看下吧
2016-08-08
Spring Cloud体系实现标签路由的方法示例
这篇文章主要介绍了Spring Cloud体系实现标签路由的方法示例，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2019-05-05
通过原理解析Spring mvc的内置编码过滤器
这篇文章通过原理主要给大家解析了关于Spring mvc的内置编码过滤器的相关资料，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面来一起看看吧。
2017-09-09
MyBatis动态SQL foreach标签实现批量插入的方法示例
这篇文章主要介绍了MyBatis动态SQL foreach标签实现批量插入的方法示例，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-06-06
java如何根据HttpServletRequest获取IP地址
文章介绍了几种代理服务器转发服务请求头的方法,这些请求头可能包含真实IP地址,但并不是所有的代理都会包括这些请求头,而且这些IP地址可能被伪造
2025-03-03
Java中的Semaphore信号量详解
这篇文章主要介绍了Java中的Semaphore信号量详解,Semaphore（信号量）是用来控制同时访问特定资源的线程数量,通过协调各个线程以保证合理地使用公共资源,需要的朋友可以参考下
2023-12-12
Java gRPC拦截器简单实现分布式日志链路追踪器过程详解
有请求的发送、处理，当然就会有拦截器的需求，例如在服务端通过拦截器统一进行请求认证等操作，这些就需要拦截器来完成，今天松哥先和小伙伴们来聊一聊gRPC中拦截器的基本用法，后面我再整一篇文章和小伙伴们做一个基于拦截器实现的JWT认证的gRPC
2023-03-03