并发通信、生产者消费者模型

多进程之间通信的限制

看一个例子:

import multiprocessing as mp
data=666
def func():
    global data
    data=222
p=mp.Process(target=func)
p.start()
p.join()
print(data)

>>>666

可以看到,声明为global的data也没有发生变化,输出结果仍然是666,这正是多进程之间通信的限制,各个进程之间是相互独立的,互不干扰的内存空间。因此如果想要空想数据就必须开辟一段共享的内存空间。就要用到Manger对象。

Manger对象

我们常用的Manger对象空间有list(),dict(),Queue()三种,下面举一个List()的简单例子。

from multiprocessing import Process,Manager
mgr=Manager()            #创建服务器进程并返回通信的管理器
list_proxy=mgr.list()    #通过管理器在列表中开辟空间,并返回一个代理
print(list_proxy)
def func(list_ex):
    list_ex.append(‘a‘)#把代理传给子进程子进程就可以通过代理来访问共享的内存空间了。
p=Process(target=func,args=(list_proxy,))
p.start()
p.join()
print(list_proxy)

>>>[]
   [‘a‘]

线程间的共享与同步锁

进程间如果不通过Manger对象是无法进行内存共享的,那么对于线程呢?对于Python来讲每一次只能执行一个线程,由于GIL锁的存在。我们来看例子。

import threading
data=666
def func():
    global data
    data=222

t=threading.Thread(target=func)
t.start()
t.join()
print(data)

>>>222

我们看到结果输出了222,也就是说全局对象更改了data的值,由此可见线程之间的内存是共享的。正是因为共享的便会出现资源竞争的问题,我们来看例子:

import threading
data=0
n=10000000     #这个n必须足够大才能看出效果
def add(n):
    global data
    for i in range(n):
        data+=1

def sub(n):
    global data
    for i in range(n):
        data-=1

a=threading.Thread(target=add,args=(n,))
s=threading.Thread(target=sub,args=(n,))
a.start()
s.start()
a.join()
s.join()
print(data)

>>>-1561473

可以看到本来应该为0的值,在基数足够大的时候就出现了问题,这就是由于线程之间的内存共享导致的,所以为了解决这一个问题就出现了同步锁的概念,说白了就是加上锁,然后控制资源的访问权限这样就会避免资源竞争的出现。看代码。

import threading

lock=threading.Lock()    #生成一把锁
data=0
n=10000000
def add(n):
    global data
    for i in range(n):
        lock.acquire()
        data+=1
        lock.release()

def sub(n):
    global data
    for i in range(n):
        lock.acquire()
        data-=1
        lock.release()

a=threading.Thread(target=add,args=(n,))
s=threading.Thread(target=sub,args=(n,))
a.start()
s.start()
a.join()
s.join()
print(data)

>>>0

这样通过锁来访问就正确的得出结果了,但是要记住一点加锁之后要记得释放,或者通过with语法这样会自动帮你释放。

with lock:
    data-=1

线程与进程安全的队列

队列是一种常用的数据结构,原则是先进先出(FIFO)。

线程安全的队列

主要方法包括:

  • 入队:put(item)
  • 出队:get()
  • 测试空:empty() #近似
  • 测试满:full() #近似
  • 队列长度:qsize() #近似
  • 任务结束:task_done()
  • 等待完成:join()

进程安全队列

进程的队列要用到之前提到的Manger对象,mgr.Queue()

主要方法包括:

  • 入队:put(item)
  • 出队:get()
  • 测试空:empty() #近似
  • 测试满:full() #近似
  • 队列长度:qsize() #近似

例子我们放到下面的生产者消费者模型中讲解。

生产者消费者模型

何所谓生产者消费者模型?

就是说我们把进程之间的通信分开考虑,生产者只要往队列里面丢东西,消费者只要从队列里取东西,而二者不用考虑对方。

多线程实现

#生产者消费者模型
import queue
import threading
import random
import time

class Producer(threading.Thread):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def run(self):
        while True:
            #生成了一个数据
            data = random.randint(0, 99)
            self.queue.put(data)   #把数据丢进队列中
            print(‘生产者: 生产了:‘, data)
            time.sleep(1)

class Concumer(threading.Thread):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def run(self):
        while True:
            item = self.queue.get() #从队列中拿一个数据
            print(‘消费者: 从队列中拿到:‘, item)

q = queue.Queue(5)  #创建一个队列
producer = Producer(q)  #创建一个生产者
concumer = Concumer(q)  #创建一个消费者

producer.start()
concumer.start()

>>>生产者: 生产了: 46
消费者: 从队列中拿到: 46
生产者: 生产了: 9
消费者: 从队列中拿到: 9
生产者: 生产了: 39
消费者: 从队列中拿到: 39
生产者: 生产了: 89
消费者: 从队列中拿到: 89

多进程实现

import multiprocessing
import random
import time

class Producer(multiprocessing.Process):
    def __init__(self,queue):
        super().__init__()
        self.queue=queue
    def run(self):
        while True:
            data=random.randint(0,100)
            self.queue.put(data)
            print("生产者生产了数据{}".format(data))
            time.sleep(1)

class Consumer(multiprocessing.Process):
    def __init__(self,queue):
        super().__init__()
        self.queue=queue
    def run(self):
        while True:
            item=self.queue.get()
            print("消费者消费{}".format(item))
if __name__ == ‘__main__‘:
    manger = multiprocessing.Manager()
    queue_m = manger.Queue()
    producer=Producer(queue_m)
    consumer=Consumer(queue_m)
    producer.start()
    consumer.start()
    producer.join()
    consumer.join()

>>>生产者生产了数据20
消费者消费20
生产者生产了数据62
消费者消费62
生产者生产了数据26
消费者消费26
生产者生产了数据36
消费者消费36
生产者生产了数据56
消费者消费56

原文地址:https://www.cnblogs.com/austinjoe/p/9685970.html

时间: 2025-01-06 10:31:57

并发通信、生产者消费者模型的相关文章

Java线程:并发协作-生产者消费者模型

对于多线程程序来说,不管任何编程语言,生产者消费者模型都是最经典的. 实际上,准确的说应该是"生产者-消费者-仓储"模型,离开了仓储,生产者消费者模型就显得没有说服力了. 对于此模型,应该明确以下几点: 生产者仅仅在仓储未满时候生产,仓满则停止生产. 消费者仅仅在仓储有产品时候才能消费,仓空则等待. 当消费者发现仓储没有产品的时候会通知生产者生产. 生产者在生产出可消费产品时候,应该通知消费者去消费. 此模型将要结合java.lang.Object的wait与notify,notify

c++并发练习---生产者消费者模型

问题:有一个生产者,多个消费者,生产者每生产一个,放入队列,多个消费者顺序从队列中取出数据,打印最终结果. 分析:首先这题,我本意应该设计成如下模型:生产者单开一个线程,向队列中放入数据,而消费者在锁的保护下,从队列中去数据.但是在实际编程中,发现在队列只有100个数的情况,线程不切换,当队列数据多的时候,会发生切换,但是也不是我所想象的那种随机切换,思考至今,也没有一个合理的解释/(ㄒoㄒ)/~~.最后我把题目改成了生产者没生产一个数据,就通知消费者去取,这样保证了一定的同步,但是估计或许开销

【Windows】用信号量实现生产者-消费者模型

线程并发的生产者-消费者模型: 1.两个进程对同一个内存资源进行操作,一个是生产者,一个是消费者. 2.生产者往共享内存资源填充数据,如果区域满,则等待消费者消费数据. 3.消费者从共享内存资源取数据,如果区域空,则等待生产者填充数据. 4.生产者的填充数据行为和消费者的消费数据行为不可在同一时间发生. 下面用Windows的信号量以及线程等API模拟生产者-消费者模型 #include <Windows.h> #include <stdio.h> #define N 100 #d

python并发编程之多进程(二):互斥锁(同步锁)&amp;进程其他属性&amp;进程间通信(queue)&amp;生产者消费者模型

一,互斥锁,同步锁 进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的, 竞争带来的结果就是错乱,如何控制,就是加锁处理 part1:多个进程共享同一打印终端 #并发运行,效率高,但竞争同一打印终端,带来了打印错乱 from multiprocessing import Process import os,time def work(): print('%s is running' %os.getpid()) time.sleep(2) print('

python—day29 守护进程、互斥锁、模拟抢票、IPC通信机制、生产者消费者模型

1.守护进程: 什么是守护进程,假如你是皇帝,每日每夜守护你的就是太监,守护进程就相当于太监,当皇帝驾崩后太监也需要陪葬,所以守护进程当父进程销毁时就一起销毁: 1 from multiprocessing import Process 2 3 import time 4 5 def task(name): 6 7 time.sleep(0.5) 8 print('%s' %name) 9 10 11 if __name__ == '__main__': 12 p = Process(targe

Python之路(第三十八篇) 并发编程:进程同步锁/互斥锁、信号量、事件、队列、生产者消费者模型

一.进程锁(同步锁/互斥锁) 进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的, 而共享带来的是竞争,竞争带来的结果就是错乱,如何控制,就是加锁处理. 例子 #并发运行,效率高,但竞争同一打印终端,带来了打印错乱 from multiprocessing import Process import os,time def work(): print('%s is running' %os.getpid()) time.sleep(2) print('

[Java并发编程实战] 阻塞队列 BlockingQueue(含代码,生产者-消费者模型)

见贤思齐焉,见不贤而内自省也.-<论语> PS: 如果觉得本文有用的话,请帮忙点赞,留言评论支持一下哦,您的支持是我最大的动力!谢谢啦~ Java5.0 增加了两种新的容器类型,它们是指:Queue 和 BlockingQueue.Queue 用来临时保存一组等待处理的元素.BlockingQueue 扩张了 Queue 接口,增加了可阻塞的插入和获取等操作. BlockingQueue 通常运用于一个线程生产对象放入队列,另一个线程从队列获取对象并消费,这是典型的生产者消费者模型. 这里写图

5 并发编程--队列&amp;生产者消费者模型

1.队列的介绍 进程彼此之间互相隔离,要实现进程间通信(IPC),multiprocessing模块支持两种形式:队列和管道,这两种方式都是使用消息传递的 创建队列的类(底层就是以管道和锁定的方式实现): Queue([maxsize]):创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue实现多进程之间的数据传递. 参数介绍: maxsize是队列中允许最大项数,省略则无大小限制. 但需要明确: 1.队列内存放的是消息而非大数据 2.队列占用的是内存空间,因而maxsize即便

11.python并发入门(part8 基于线程队列实现生产者消费者模型)

一.什么是生产者消费者模型? 生产者就是生产数据的线程,消费者指的就是消费数据的线程. 在多线程开发过程中,生产者的速度比消费者的速度快,那么生产者就必须等待消费者把数据处理完,生产者才会产生新的数据,相对的,如果消费者处理数据的速度大于生产者,那么消费者就必须等待生产者. 为了解决这种问题,就有了生产者消费者模型. 生产者与消费者模型,是通过一个容器,来解决生产者和消费者之间的耦合性问题,生产者和消费者之间并不会直接通信,这样生产者就无需等待消费者处理完数据,生产者可以直接把数据扔给队列,这个