百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术文章 > 正文

通过生产者消费者问题学习多线程

wuantov 2025-08-05 21:53 6 浏览

生产者消费者问题,也称有限缓冲问题,是多线程同步问题的经典案例。该问题描述了两个共享固定大小缓冲区的线程——即所谓的“生产者”和“消费者”——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中,然后重复此过程。与此同时,消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据,消费者也不会在缓冲区中空时消耗数据。



要理解生产消费者问题,首先应弄清PV操作与信号量。信号量的值与相应资源的使用情况有关。当它的值大于0时,表示当前可用资源的数量;当它的值小于0时,其绝对值表示等待使用该资源的线程个数,信号量的值仅能由PV操作来改变。一般来说,信号量S >= 0时,S表示可用资源的数量。执行一次P操作意味着请求分配一个单位资源,因此S的值减1;当S < 0时,表示已经没有可用资源,请求者必须等待别的进程释放该类资源,它才能运行下去。而执行一个V操作意味着释放一个单位资源,因此S的值加1;若S < 0,表示有某些进程正在等待该资源,因此要唤醒一个等待状态的进程,使之运行下去。

正确的实现生产者消费者问题,一般需要三个信号量:

  • empty,表示缓冲区中的空闲空间,即还可以容纳多少元素,初始值为缓冲区大小。
  • full,表示缓冲区中已经有多少元素,初始值为0。
  • mutex,互斥信号量,初始值为1。对缓冲区的访问是排它的,即某一时刻只能有一个生产者向缓冲区里放元素,或一个消费者从中取元素。

生产者

消费者

P(empty)
P(mutex)
向缓冲区增加元素
V(mutex)
V(full)

P(empty)
P(mutex)
向缓冲区增加元素
V(mutex)
V(full)

使用内置锁

生产者、消费者的问题,其实就是线程间同步的问题,在java中,是通过synchronizedwaitnotifyAll来实现的。

public class Buffer<V> {
    private List<V> list = new LinkedList<V>();
    private int size;

    public Buffer(int size) {
        this.size = size;
    }

    public void add(V v) throws InterruptedException {
        synchronized(this) {
            while (list.size() == size) {
                this.wait();
            }


            list.add(v);
            this.notifyAll();
        }
    }

    public V take() throws InterruptedException {
        synchronized(this) {
            while (list.size() == 0) {
                this.wait();
            }

            V v = list.get(0);
            list.remove(v);
            this.notifyAll();

            return v;
        }
    }
}

add方法中对list.size() == size的判断相当于P(empty)操作,当缓冲区中没有空闲位置的时候通过调用wait方法强制阻塞当前线程同时释放this对象的锁,使得别的线程(消费者)有机会获得锁去取走元素。

list.size() == size的判断放进while中判断而不是if来判断是因为可能有多个生产者,当它们被唤醒后只有其中一个能真正向缓冲区增加元素,其它生产者会因为竞争失败再次处于阻塞状态;如果只有一个生产者线程,是可以使用if替换while的。

当生产者向缓冲区中增加元素后,通过this.notifyAll()方法通知消者费线程到缓冲区中去取元素。这里需要特别说明,在this对象上等待的线程,即可能是等待从缓冲区取的消费者线程,也可能是等待向缓冲区放的生产者。当调用this.notifyAll方法时,所有这些等待的线程都会被唤醒,某种程度上来讲,这种性能的损耗很大,第一:唤醒生产者线程完全没有意义;第二:被唤醒的消费者线程中只有一个能真正得到执行。另外,不难看出如果有多个生产者与消费者线程的情况下,使用notify()方法代替notifyAll()方法有可能发生活锁的情况。

条件通知


调用notifyAll()方法的代价很高,其实我们是可以减少它的调用次数的。想想add方法中的notifyAll()方法的本意是想通知阻塞的消费者线程去缓冲区取元素,如果在生产者向缓冲区里增加元素时,缓冲区并不是空的,意味着此刻并不会有消费者因为缓冲区为空而等待,也就无需进行通知了。

采用条件通知的方式重新实现的add()方法如下。

public void add(V v) throws InterruptedException {
    synchronized(this) {
        int currentSize = list.size();
        while (currentSize == size) {
            this.wait();
        }

        list.add(v);

        if (0 == currentSize) {
            this.notifyAll();
        }
    }
}

使用显示锁

使用内置锁有一个明显的缺陷,在像生产者、消费者这类问题上,多个线程可能在同一个条件队列上等待不同的条件谓词,如生产者向缓冲区增加元素后调用notifyAll()方法时,其它的生产者线程也会被唤醒。

另一方面,诸多被唤醒的消费者线程中也只有一个能真正得到执行。使用Lock与Condition可以使得等待特定条件谓词的线程处在同一个条件队列,从根本上解决了内置锁的缺陷。

使用显示锁重新实现的Buffer如下:

public class Buffer<V> {
    private List<V> list = new LinkedList<V>();
    private Lock lock = new ReentrantLock();
    private Condition full = lock.newCondition();
    private Condition empty= lock.newCondition();

    private int size;

    public Buffer(int size) {
        this.size = size;
    }

    public void add(V v) throws InterruptedException {
        lock.lock();
        try {
            while (list.size() == size) {
                full.await();
            }

            list.add(v);
            empty.signal();
        } finally {
            lock.unlock();
        }
    }

    public V take() throws InterruptedException {
        lock.lock();
        try {
            while (list.isEmpty()) {
                empty.await();
            }

            V v = list.get(0);
            list.remove(v);
            full.signal();

            return v;
        } finally {
            lock.unlock();
        }
    }
}

缓冲区是否已满和缓冲区是否为空分别由fullempty两个条件谓词表示,因而在唤醒的时候具有更强的针对性,而且只需要唤醒其中一个等待的线程(使用signal而不是signalAll),在竞争激烈的情况下可以减少很多无效的线程上下文切换与加锁操作从而避免性能损耗。

内置锁与显示锁


Lock对象还提供了tryLocklockInterruptibly等方法,支持加锁时设置超时和响应中断等能力,比内置锁灵活很多。在jdk1.5中,显示锁的性能是明显好于内置锁的;在jdk1.6以后的版本使用了改进的算法管理内置锁,两者之间的性能差异几乎可以忽略。

内置锁虽然在灵活性方面差了些,但是有一个明显的优势:在线程转储中可以看到哪些调用帧中获得了哪些锁,对于分析死锁等问题时带来的便利是显示锁无法比拟的。另外,未来提升内置锁性能的可能性应该会更高,因为它是JVM的内置属性,除非需要处理一些内置锁无法满足的问题,否则还是应该优先使用内置锁。

4. 使用concurrent包中的工具类

Semaphore

jdk1.5引入了Semaphore类,表示信号量,accqure方法相当于P操作,release方法相当于V操作。在掌握了生产者、消费者的解题流程后,使用Semaphore来实现Buffer也是很简单的事,使用信号量实现的Buffer如下:

public class Buffer<V> {
    private volatile List<V> list = new LinkedList<V>();
    private Semaphore mutex = new Semaphore(1);
    private Semaphore full;
    private Semaphore empty = new Semaphore(0);

    public Buffer(int size) {
        full  = new Semaphore(size);
    }

    public void add(V v) throws InterruptedException {
        full.acquire();

        try {
            mutex.acquire();
            list.add(v);
        } catch(InterruptedException e) {
            full.release();
        } finally {
            mutex.release();
        }

        empty.release();
    }

    public V take() throws InterruptedException {
        empty.acquire();

        try {
            mutex.acquire();
            V v = list.get(0);
            list.remove(v);

            return v;
        } catch(InterruptedException e) {
            empty.release();
            throw e;
        } finally {
            mutex.release();
            full.release();
        }
    }
}

当然了,这种方式实现的Buffer无论是性能,又或者可读性方面都不比前两种方式好,仅当作练习的话那还是很经典的一个例子的。

BlockingQueue

jdk1.5中新引入的BlockingQueue,它是专门为解决生产者、消费者问题而生的,它定义了四组方法以及五种不同的实现:

  • ArrayBlockingQueue
  • LinkedBlockingQueue
  • DelayQueue
  • PriorityBlockingQueue
  • SynchronousQueue

使用LockFree算法实现

在竞争的情况下系统的性能会因为加锁产生上下文切换与调度延迟而降低,而非竞争的情况下多余的加锁操作本身也会消耗掉一部分性能。

Compare And Swap

CAS 指的是现代 CPU 广泛支持的一种对内存中的共享数据进行操作的一种特殊指令。这个指令会对内存中的共享数据做原子的读写操作。简单介绍一下这个指令的操作过程:首先,CPU 会将内存中将要被更改的数据与期望的值做比较。然后,当这两个值相等时,CPU 才会将内存中的数值替换为新的值。否则便不做操作。最后,CPU 会将旧的数值返回。这一系列的操作是原子的。它们虽然看似复杂,但却是 Java 5 并发机制优于原有锁机制的根本。简单来说,CAS 的含义是“我认为原有的值应该是什么,如果是,则将原有的值更新为新值,否则不做修改,并告诉我原来的值是多少”。


java.util.concurrent.atomic包提供了大部分数据类型的原子封装,在原有数据类型的基础上,提供了原子性的操作方法,保证了线程安全。下面是采用CAS方式实现的Buffer:

public class Buffer<E> {
    private static class Node<E> {
        E item;
        AtomicReference<Node<E>> next;

        Node(E item, Node<E> next) {
            this.item = item;
            this.next = new AtomicReference<Node<E>>(next);
        }
    }

    private final Node<E> dummy = new Node<E>(null, null);
    private final AtomicReference<Node<E>> head = new AtomicReference<Node<E>>(dummy);
    private final AtomicReference<Node<E>> tail = new AtomicReference<Node<E>>(dummy);

    public boolean add(E e) {
        Node<E> newNode = new Node<E>(e, null);
        while (true) {
            Node<E> t = tail.get();
            Node<E> residue = t.next.get();

            if (t == tail.get()) {
                if (residue == null) {
                    if (t.next.compareAndSet(null, newNode)) {
                        tail.compareAndSet(t, newNode);
                        return true;
                    }
                } else {
                    tail.compareAndSet(t, residue);
                }
            }
        }
    }

    public E take() {
        while (true) {
            Node<E> h = head.get();
            Node<E> t = tail.get();
            Node<E> first = h.next.get();

            if (h == head.get()) {
                if (h == t) {
                    if (first == null) {
                        return null;
                    } else {
                        tail.compareAndSet(t, first);
                    }
                } else if (head.compareAndSet(h, first)) {
                    E e = first.item;
                    if (e != null) {
                        first.item = null;
                        return e;
                    }
                }
            }
        }
    }
}

当竞争程度不高时,基于CAS的实现在性能上远远超过基于锁的实现,但是在竞争激烈的情况下,CAS的性能会比锁定的方式差很多,因为CAS是通过不断地重试、回退的方式处理竞争的,在竞争激烈的情况下会消耗很多的CPU资源;CAS的另一个缺点是会引发ABA问题。理解CAS方式实现的Buffer是有一定的困难的,如果您对非阻塞算法感兴趣,请关注笔者后续的文章。

相关推荐

UV 工具深度解析与实践指南

一个工具可以替代pip、pip-tools、pipx、poetry、pyenv、twine、virtualenv等一、工具概述1.1核心定位UV是由Rust编写的新一代Python生态...

MCP 官方文档:开发你自己的MCP —— MCP 天气服务

一、MCP简介MCP是一个开放协议,它为应用程序向LLM提供上下文的方式进行了标准化。你可以将MCP想象成AI应用程序的USB-C接口。就像USB-C为设备连接各种外设和配件提...

一文使你彻底学会FastAPI

什么是API?想象一下,你在一家餐馆,你想点一些食物。为了得到你想要的食物,你和服务员交谈,告诉他们你的订单。然后,服务员将您的订单带到厨房,厨师会根据您的要求准备食物。最后,服务员把你的食物带回...

手把手教你进行Python虚拟环境配置

/1前言/咱们今天就来说一下Python的虚拟环境,可能有的小伙伴会疑惑,Python的虚拟环境有什么用呢?接下来我们一起来探讨一下。/2虚拟环境的作用/咱们今天就来说一下Python的虚拟环...

LangGraph集成DeepSeek实现AI对话

本文介绍如何使用AI开源框架LangGraph,集成调用DeepSeek大模型,实现AI对话功能,并给出Python示例代码。一、LangGrap是什么LangGrap:LangGrap是一个用于构...

一文带你掌握Python自带venv虚拟环境

前边几篇文章,我们介绍了如何使用conda来管理python的虚拟环境。这时可能有人会有疑问,既然这个功能这么有用,那有官方提供的解决方案吗?其实是有的,在python3.3版本及以后,...

Pytorch详细安装过程

1、安装anaconda官网(https://www.anaconda.com/products/distribution#Downloads)下载,使用管理员身份运行(不使用似乎也没事)这里选择Ju...

Python简介与开发环境搭建详细教程

1.1Python简介与开发环境搭建详细教程一、Python语言简介1.Python的核心特点2.Python的应用领域表1.1Python主要应用领域领域典型应用常用库Web开发网站后端D...

Python开发中的虚拟环境管理

Python开发中,虚拟环境管理帮助隔离项目依赖,避免不同项目之间的依赖冲突。虚拟环境的作用隔离依赖:不同项目可能需要不同版本的库,虚拟环境可以为每个项目创建独立的环境。避免全局污染:全局安装的库可...

Python项目创建全流程指南

以下是创建Python项目的超详细步骤指南,涵盖从环境配置到项目部署的全流程:一、环境准备安装PythonO官网下载:访问python.org下载最新稳定版O验证安装:终端输入pyth...

LangChain开发MCP Server和MCP Client

本文介绍了如何通过LangChain实现MCP调用。通过模拟一个简单的算术计算器,基于MCPServer运行,并使用MCPClient进行调用。最终,通过集成DeepSeek大...

怎么给虚拟环境安装pdfplumber

1.找到虚拟环境激活脚本位置你的虚拟环境在C:\Users\shiqi\PycharmProjects\pythonProject2\.venv路径下,激活脚本Activate.ps1(Pow...

Python调试器实现断点系统

在软件开发过程中,调试是不可或缺的重要环节。断点系统作为调试器的核心功能,允许开发者在程序执行的特定位置暂停程序运行,检查变量状态、分析程序流程,从而快速定位和解决问题。断点系统原理断点系统的实现基于...

Python从放弃到入门:公众号历史文章爬取为例谈快速学习技能

这篇文章不谈江流所专研的营销与运营,而聊一聊技能学习之路,聊一聊Python这门最简单的编程语言该如何学习,我完成的第一个Python项目,将任意公众号的所有历史文章导出成PDF电子书。或许我这个Py...

能跑源码,还提供数据集:这里有一个入门企业级验证码识别项目

机器之心专栏作者:kerlomz网上关于验证码识别的开源项目众多,但大多是学术型文章或者仅仅是一个测试demo,那么企业级的验证码识别究竟是怎样的呢?1.前言网上关于验证么识别的开源项目众多,但大...