python之进程----Queue

一、Queue是通过multiprocessing使用

from multiprocessing import Process,Queue

import time

import random

import os

def consumer(q):

    while True:

        res=q.get()

        if res is None:

            break

        time.sleep(random.randint(1,3))

        print('\033[45m%s 吃了 %s\033[0m' % (os.getpid(), res))

def producer(q):

    for i in range(5):

        time.sleep(2)

        res='包子%s' %i

        q.put(res)

        print('\033[44m%s 制造了 %s\033[0m' %(os.getpid(),res))

    q.put(None)

if __name__ == '__main__':

    q=Queue()

    #生产者们：厨师们

    p1=Process(target=producer,args=(q,))

    #消费者们：吃货们

    p2=Process(target=consumer,args=(q,))

    p1.start()

    p2.start()

    p1.join()

    p2.join()

    print('主')

生产者，消费者模型1

from multiprocessing import Process,Queue

import time

import random

import os

def consumer(q):

    while True:

        res=q.get()

        if res is None:break

        time.sleep(random.randint(1,3))

        print('\033[45m%s 吃了 %s\033[0m' % (os.getpid(), res))

def product_baozi(q):

    for i in range(3):

        time.sleep(2)

        res='包子%s' %i

        q.put(res)

        print('\033[44m%s 制造了 %s\033[0m' %(os.getpid(),res))

def product_jiaozi(q):

    for i in range(3):

        time.sleep(2)

        res='饺子%s' %i

        q.put(res)

        print('\033[44m%s 制造了 %s\033[0m' %(os.getpid(),res))

def product_dabing(q):

    for i in range(3):

        time.sleep(2)

        res='大饼%s' %i

        q.put(res)

        print('\033[44m%s 制造了 %s\033[0m' %(os.getpid(),res))

if __name__ == '__main__':

    q=Queue()

    #生产者们：厨师们

    p1=Process(target=product_baozi,args=(q,))

    p2=Process(target=product_jiaozi,args=(q,))

    p3=Process(target=product_dabing,args=(q,))

    #消费者们：吃货们

    p4=Process(target=consumer,args=(q,))

    p5=Process(target=consumer,args=(q,))

    p_l=[p1,p2,p3,p4,p5]

    for p in p_l:

        p.start()

    # for p in p_l:

    #     p.join()

    # p1.start()

    # p2.start()

    # p3.start()

    # p4.start()

    # p5.start()

    p1.join()

    p2.join()

    p3.join()

    q.put(None)

    q.put(None)

    p4.join()

    p5.join()

    print('主')

生产者，消费者模型2

  q .put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），
 并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。
 如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。

 q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。
如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。
如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常.

1:可以往队列里放任意类型的数据 2 队列：先进先出

from multiprocessing import Process,Queue

q=Queue(3)

q.put('first')

q.put('second')

q.put('third')

# q.put('fourht')

print(q.get())

print(q.get())

print(q.get())

# print(q.get())

q.put和q.get

from multiprocessing import Process,Queue

q=Queue(3)

q.put('first',block=False)

q.put('second',block=False)

q.put('third',block=False)

# q.put('fourth',block=False)

q.put('fourth',block=True,timeout=3)

q.get(block=False)

q.get(block=True,timeout=3)

q.get_nowait() #q.get(block=False)

p.get的参数

二、JoinableQueue同样通过multiprocessing使用。

JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：

q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常

q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

from multiprocessing import Process,JoinableQueue

import time

import random

import os

def consumer(q):

    while True:

        res=q.get()

        time.sleep(random.randint(1,3))

        print('\033[45m%s 吃了 %s\033[0m' % (os.getpid(), res))

        q.task_done()

def product_baozi(q):

    for i in range(5):

        time.sleep(2)

        res='包子%s' %i

        q.put(res)

        print('\033[44m%s 制造了 %s\033[0m' %(os.getpid(),res))

    q.join()

if __name__ == '__main__':

    q=JoinableQueue()

    #生产者们：厨师们

    p1=Process(target=product_baozi,args=(q,))

    #消费者们：吃货们

    p4=Process(target=consumer,args=(q,))

    p4.daemon=True

    p1.start()

    p4.start()

    p1.join()

    print('主')

生产者，消费者模型3

from multiprocessing import Process,JoinableQueue

import time

import random

import os

def product_baozi(q):

    for i in range(3):

        time.sleep(2)

        res='包子%s' %i

        q.put(res)

        print('\033[44m%s 制造了 %s\033[0m' %(os.getpid(),res))

    q.join()

def product_jiaozi(q):

    for i in range(3):

        time.sleep(2)

        res='饺子%s' %i

        q.put(res)

        print('\033[44m%s 制造了 %s\033[0m' %(os.getpid(),res))

    q.join()

def product_dabing(q):

    for i in range(3):

        time.sleep(2)

        res='大饼%s' %i

        q.put(res)

        print('\033[44m%s 制造了 %s\033[0m' %(os.getpid(),res))

    q.join()

def consumer(q):

    while True:

        res = q.get()

        time.sleep(random.randint(1, 3))

        print('\033[45m%s 吃了 %s\033[0m' % (os.getpid(), res))

        q.task_done()

if __name__ == '__main__':

    q=JoinableQueue()

    #生产者们：厨师们

    p1=Process(target=product_baozi,args=(q,))

    p2=Process(target=product_jiaozi,args=(q,))

    p3=Process(target=product_dabing,args=(q,))

    #消费者们：吃货们

    p4=Process(target=consumer,args=(q,))

    p5=Process(target=consumer,args=(q,))

    p4.daemon=True

    p5.daemon=True

    p_l=[p1,p2,p3,p4,p5]

    for p in p_l:

        p.start()

    p1.join()

    p2.join()

    p3.join()

    print('主')

生产者，消费者模型4

三，互斥锁

互斥锁应用：

from multiprocessing import Process,Lock

import os

import time

def work(mutex):

    mutex.acquire()

    print('task[%s] 上厕所' %os.getpid())

    time.sleep(3)

    print('task[%s] 上完厕所' %os.getpid())

    mutex.release()

if __name__ == '__main__':

    mutex=Lock()

    p1=Process(target=work,args=(mutex,))

    p2=Process(target=work,args=(mutex,))

    p3=Process(target=work,args=(mutex,))

    p1.start()

    p2.start()

    p3.start()

    print('主')

互斥锁

四、模拟抢票：

from multiprocessing import Process,Lock

import json

import time

import random

import os

def search():

    dic=json.load(open('db.txt',))

    print('剩余票数%s' %dic['count'])

def get_ticket():

    dic=json.load(open('db.txt',))

    if dic['count'] > 0:

        dic['count']-=1

        json.dump(dic,open('db.txt','w'))

        print('%s 购票成功' %os.getpid())

def task(mutex):

    search()

    time.sleep(random.randint(1, 3)) #模拟购票一系列繁琐的过程所花费的时间

    mutex.acquire()

    get_ticket()

    mutex.release()

if __name__ == '__main__':

    mutex=Lock()

    for i in range(50):

        p=Process(target=task,args=(mutex,))

        p.start()

模拟抢票

五、process对象的其他属性补充

from multiprocessing import Process

import os

import time

def work():

    print('%s is working' %os.getpid())

    time.sleep(3)

if __name__ == '__main__':

    p1=Process(target=work)

    p2=Process(target=work)

    p3=Process(target=work)

    p1.daemon=True

    p2.daemon=True

    p3.daemon=True

    p1.start() #初始化1

    p2.start() #初始化2

    p3.start() #初始化3

    p3.join()

    p1.join()

    p2.join()

    print('基于初始化的结果来继续运行')

process属性补充1

from multiprocessing import Process

import os

import time

def work():

    print('%s is working' %os.getpid())

    time.sleep(3)

if __name__ == '__main__':

    p1=Process(target=work)

    # p2=Process(target=work)

    # p3=Process(target=work)

    p1.start() #初始化1

    # p2.start() #初始化2

    # p3.start() #初始化3

    # p1.terminate()

    # time.sleep(3)

    # print(p1.is_alive())

    print(p1.name)

    print(p1.pid)

    print('基于初始化的结果来继续运行')

process属性补充2

巴特西

python之进程----Queue

最新文章

热门文章