python多进程并发和多线程并发和协程

为什么需要并发编程?
如果程序中包含I/O操作,程序会有很高的延迟,CPU会处于等待状态,这样会浪费系统资源,浪费时间
1.Python的并发编程分为多进程并发和多线程并发
多进程并发:运行多个独立的程序,优势在于并发处理的任务都有操作系统管理,不足的是程序和各个进程间通信和数据共享不方便
多线程并发:有程序员管理并发处理人物,这种并发的可以方便的在线程间共享数据,前提是不能被锁住
对于计算密集型程序:多进程并发优于多线程并发,计算密集型指的是:程序运行的时间大部分都消耗在cpu的运算处理过程中,而对内存磁盘的消耗时间较短对于i/o密集型程序:多线程并发优于多进程并发,i/o密集型与计算密集型正好相反
2.python支持的多进程并发有两种方式
1.通过进程安全的数据结构:multiprocess.JoinableQueue:这种数据结构程序员自己管理'加锁'过程,不用担心死锁问题
2.通过concurr.futures抽象出来的ProcessPollExecutor
multiprocess的JoinableQueue
multiprocess(是进程间安全的) 是Python标准库中的支持进程的模块,JoinableQueue队列本质上是FIFO队列,
与一般的队列(queue中的Queue)区别在于:JoinableQueue是能支持多进程并发和保证进程间的数据通信,
是进程间安全的,这意味着我们不用担心它的互斥和死锁问题,JoinableQueue主要可以用来存放执行任务和收集任务的执行结果
JoinableQueue队列为什么是进程间安全的?
    因为不管什么队列,都是自带了锁,所中的上下文管理中(with)包含了acquire()和release()方法。
    队列基于文件家族的socket实现的,队列中是有管道(pipe)和锁实现的。
from multiprocessing import JoinableQueue,Process,current_process
def consumer(jq,name):
    while True:
        word = jq.get()
        print('%s 取到了%s'%(name,word))
        jq.task_done()
        # print(jq.task_done())

def print_word(jq,produce_name):
    for c in [chr(ord('A')+i) for i in range(26)]:
        jq.put(c)
        print("%s 生产了一个 %s"%(produce_name,c))
    jq.join()

if __name__ == '__main__':
    jq = JoinableQueue()   #不管用什么队列,都必须要先实例化
    produce_name = 'admin'
    # consumer_list = ['kobe','t-mac']
    pn = Process(target=print_word,args=(jq,produce_name))
    tt1 = Process(target=consumer, args=(jq,'kobe'))
    tt2 = Process(target=consumer, args=(jq,'t-mac'))
    tt1.daemon =True
    tt2.daemon =True
    tt1.start()
    tt2.start()
    pn.start()
    pn.join()

'''
使用JoinableQueue实现多线程搞得时候要注意这几个方法的关联
1.pn.join() 为什么子线程pn要设置阻塞?
    首先要知道join方法是Process提供来管理子进程的,在同步阻塞的时候,只关心子线程是否执行完毕,只要子线程结束,才会执行阻塞后的代码
2.jq.join()为什么子进程设置阻塞?
    因为jq要等待consumer方法将全部的字母取走,只要consumer将全部的字母取走之后,jq才不阻塞了,代表着print_word方法已经将工作做完了
    jq.join()在这里可以理解成:当队列为空我才不阻塞了,
3.jq.task_done()是什么意思?
    通知producer里面的jq.join(),要将队列里面的计数器-1,因为有一个数据被取走了,当所有的任务处理完之后,队列的计数器为0
    也就是队列为空了,这是jq.join()就不阻塞了
4.那tt1.daemon =True和tt2.daemon =True子进程为什么要设置成守护进程?
    我们子啊consumer里面写的是while True,理解是这个死循环永远不能退出
    但是守护进程会随着主进程的运行完毕之后跟着结束,我们退出不了while,那把主进程kill掉,也就退出了死循环
    所以:
        pn.join()在等待这自己的子进程函数print_word执行完毕,因为执行到主线程搞得代码结束为止,所以主线程退出
        但是print_word()子线程里面的jq.join()也在阻塞这,等待jq.task_done()发过来的最后一个信号,也就是队列为空
        所以,当jq.task_done()发空信号之后jq.join()不阻塞,然后pn.join()不阻塞,程序结束
        而进程中的守护进程的特点是随着主进程的结束而结束,所以整个程序结束,name死循环也就结束了。

理解task_done():
如果进程或线程每从队列里取一次,但没有执行task_done(),则join无法判断队列到底有没有结束,在最后执行个join()是等不到结果的,会一直挂起。
可以理解为,每task_done一次 就从队列里删掉一个元素,这样在最后join的时候根据队列长度是否为零来判断队列是否结束,从而执行主线程。
'''

multiprocessing中的JoinableQueue实现多进程并发

concurrent.futures的ProcessPoolExecutor
from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
ThreadPoolExecutor 和ProcessPoolExecutor分别对threading和multiprocessing进行了高级抽象,暴露出简单的统一接口。

future 是一种对象,表示异步执行的操作。这个概念是 concurrent.futures模块和asyncio包的基础。
从Python3.4起,标准库中有两个为Future的类:concurrent.futures.Future 和 asyncio.Future。
这两个类作用相同:两个Future类的实例都表示可能已经完成或未完成的延迟计算。
Future 封装待完成的操作,可放入队列,完成的状态可以查询,得到结果(或抛出异常)后可以获取结果(或异常)。
import os,time
from urllib.request import urlopen
from concurrent.futures import ProcessPoolExecutor

def get_html(name,addr):
    ret = urlopen(addr)
    return {'name': name, 'content': ret.read()}

def print_info(connect):
    dic = connect.result()
    with open(dic['name']+'.html',mode='wb') as f:
        f.write(dic['content'])

if __name__ == '__main__':
    start = time.time()
    url_list = {
        '百度':'https://www.baidu.com',
        'p0st':'https://www.cnblogs.com/p0st/p/10453405.html',
        'jd':'https://www.jd.com',
        '博客园':'https://www.cnblogs.com',
        'a':'https://www.baidu.com',
        'b': 'https://www.cnblogs.com/p0st/p/10453405.html',
        'c': 'https://www.jd.com',
        'd': 'https://www.cnblogs.com'
    }
    p = ProcessPoolExecutor(4)
    for item in url_list:
        task = p.submit(get_html,item,url_list[item]).add_done_callback(print_info)
        task.add_done_callback(print_info)
        #使用submit向进程池中添加任务,返回的时一个future对象,使用result从中取值
        #在此过程中是并发的去执行的
        #p.submit(get_html,item,url_list[item]).add_done_callback(print_info)
        #add_done_callback 回调函数,将task
    p.shutdown()
    # 阻塞主进程 直到池中的任务都完成为止,然后进程还会回到进程池里面等待任务
    print(time.time()-start)

    #with ProcessPoolExecutor(4)  as p:
    #    for item in url_list:
    #        p.submit(get_html, item, url_list[item]).add_done_callback(print_info)
    #    p.shutdown()
    #print(time.time() - start)
'''
#ret = p.map(make, range(100))  是for循环的简便写法,一般只用于传递一个参数
#ret=返回的是返回的是生成器,再循环就可以了

#如果task.result()执行修改完了之后才会返回结果,但是如果没执行完,就会等待,所以是一个阻塞方法
#如果有可能发生阻塞的方法,那个这个方法就是阻塞方法

'''
concurrent.futures中的ProcessPoolExecutor实现多进程并发

concurrent.futures中的ProcessPoolExecutor实现多进程并发

3.python中的多线程并发也有两种方式
对于IO密集型程序,多线程并发可能要优于多进程并发。因为对于网络通信等IO密集型任务来说,决定程序效率的主要是网络延迟,这时候是使用进程还是线程就没有太大关系了。

程序与多进程基本一致,只是这里我们不必使用multiProcessing.JoinableQueue对象了,一般的队列(来自queue.Queue)就可以满足要求:
from queue import Queue
def read(q):
    while True:
        try:
            value = q.get()
            print('Get %s from queue.' % value)
            time.sleep(random.random())
        finally:
            q.task_done()

def main():
    q = queue.Queue()
    pw1 = threading.Thread(target=read, args=(q,))
    pw2 = threading.Thread(target=read, args=(q,))
    pw1.daemon = True
    pw2.daemon = True
    pw1.start()
    pw2.start()
    for c in [chr(ord('A')+i) for i in range(26)]:
        q.put(c)
    try:
        q.join()
    except KeyboardInterrupt:
        print("stopped by hand")

if __name__ == '__main__':
    main()

queuq中Queue实现多线程并发

concurrent.futures的ThreadPoolExecutor

concurrent.futures的ThreadPoolExecutor抽象接口不仅可以帮我们自动调度线程,还能:  1.主线程可以获取某一个线程(或者任务的)的状态,以及返回值。  2.当一个线程完成的时候,主线程能够立即知道。  3.让多线程和多进程的编码接口一致。
1、建立线程池:executor = ThreadPoolExecutor(max_workers= )
2、提交执行函数到线程池:task = executor.submit(func,(args))
3、获取执行结果:task.result()
4、判断线程是否完成:task.done()
5、取消还没执行的线程:task.cancel()
6、利用as_completed获取线程完成结果,返回迭代器
7、通过executor的map获取已经完成的task值
for data in executor.map(get_html,urls):
    print(data)
8、使用wait()方法阻塞线程

ThreadPoolExecutor方法

from urllib.request import  urlopen
from concurrent.futures import ThreadPoolExecutor
def get_html(name,addr):
    ret = urlopen(addr)
    return {'name': name, 'content': ret.read()}

def print_info(connect):
    dic = connect.result()
    with open(dic['name']+'.html',mode='wb') as f:
        f.write(dic['content'])
url_list = {
        '百度':'https://www.baidu.com',
        'p0st':'https://www.cnblogs.com/p0st/p/10453405.html',
        'jd':'https://www.jd.com',
        '博客园':'https://www.cnblogs.com',
        'a':'https://www.baidu.com',
        'b': 'https://www.cnblogs.com/p0st/p/10453405.html',
        'c': 'https://www.jd.com',
        'd': 'https://www.cnblogs.com'
    }
t = ThreadPoolExecutor(4)
if __name__ == '__main__':
    for item in  url_list:
        task = t.submit(get_html,item,url_list[item])    #回调函数,当执行完task后,立即将返回值传给回调函数
        task.add_done_callback(print_info)
        # 回调函数,当执行完task后,立即将返回值传给回调函数

concurrent.futures中的ThreadPoolExecutor实现多线程并发

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    print(list(executor.map(sleeper, x)))

ThreadPoolExecutor高级写法

from concurrent.futures import ProcessPoolExecutor
def pool_factorizer_go(nums, nprocs):
   nprocs=xxx
    with ProcessPoolExecutor(max_workers=nprocs) as executor:
        return {num:factors for num, factors in
                                zip(nums,
                                    executor.map(factorize_naive, nums))}

ProcessPoolExecutor高级写法

#在一个函数中等待另一个结果的完毕。
import time
def wait_on_b():
    time.sleep(5)
    print(b.result())  #b不会完成,他一直在等待a的return结果
    return 5

def wait_on_a():
    time.sleep(5)
    print(a.result())  #同理a也不会完成,他也是在等待b的结果
    return 6

executor = ThreadPoolExecutor(max_workers=2)
a = executor.submit(wait_on_b)
b = executor.submit(wait_on_a)

多线程或多进程中死锁的列子

4.进程/线程中创建协程

from multiprocessing import Process

from gevent import monkey;monkey.patch_all()
import gevent
import time
import random
class Sayhi(Process):
    def __init__(self,name):
        super().__init__()
        self.name = name

    @staticmethod
    def func(item):
        time.sleep(random.uniform(1,2))
        print('协程:sayhi %s'%item)

    def run(self):
        count = 20
        ll= []
        for item in range(count):
            gg = gevent.spawn(self.func,item)
            ll.append(gg)
        gevent.joinall(ll)

if __name__ == '__main__':
    p = Sayhi('kobe')
    p.start()
    p.join()

进程中创建协程

import os
from gevent import monkey;monkey.patch_all()
import time
import random
import gevent
from threading import Thread
class Sayhi(Thread):
    def __init__(self,name):
        super().__init__()
        self.name = name

    @staticmethod
    def func(item):
        time.sleep(random.uniform(1,2))
        print('协程:sayhi %s  %s'%(item,os.getpid()))

    def run(self):
        count = 20
        l = []
        for i in range(count):
            g = gevent.spawn(self.func,i)
            l.append(g)
        gevent.joinall(l)
if __name__ == '__main__':
    p = Sayhi('kobe')
    p.start()
    p.join()
    print(os.getpid())

线程里面开启协程

线程中创建协程

asyncino  yiled from 拉拉

ProcessPoolExecutor资料



返回系列
上一篇:Python多线程并发的误区


下一篇:Python2 socket 多线程并发 TCPServer Demo