Python 37 进程池与线程池 协程

Posted 2021-01-05 温暖你的心

一：进程池与线程池

提交任务的两种方式：

1、同步调用：提交完一个任务之后，就在原地等待，等任务完完整整地运行完毕拿到结果后，再执行下一行代码，会导致任务是串行执行

2、异步调用：提交完一个任务之后，不是原地等待，而是直接执行下一行代码，会导致任务是并发执行的，结果future对象会在任务运行完毕后自动传给回调函数

 

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time, random, os


def task(name, n):
    print(\'%s%s is running\' % (name, os.path.getpid()))
    time.sleep(random.randint(1, 3))
    return n ** 2


if __name__ == \'__main__\':
    # print(os.cpu_count())
    p = ProcessPoolExecutor(4)

    l = []
    for i in range(5):
        # 同步提交
        # res = p.submit(task, \'进程pid:\', i).result()
        # print(res)

        # 异步提交
        future=p.submit(task,\'进程pid:\',i)
        l.append(future)

    p.shutdown(wait=True) # 关闭进程池的入口，并且在原地等待进程池内所有任务运行完毕

    for future in l:
        print(future.result)
    print(\'主\')

View Code

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import time,random,os
import requests


def get(url):
    print(\'%s GET %s\' %(os.getpid(),url))
    time.sleep(3)
    response=requests.get(url)
    if response.status_code == 200:
        res=response.text
    else:
        res=\'下载失败\'
    parse(res)

def parse(res):
    time.sleep(1)
    print(\'%s 解析结果为%s\' %(os.getpid(),len(res)))

if __name__ == \'__main__\':
    urls=[
        \'https://www.baidu.com\',
        \'https://www.sina.com.cn\',
        \'https://www.tmall.com\',
        \'https://www.jd.com\',
        \'https://www.python.org\',
        \'https://www.openstack.org\',
        \'https://www.baidu.com\',
        \'https://www.baidu.com\',
        \'https://www.baidu.com\',

    ]

    p=ProcessPoolExecutor(9)
    l=[]
    start=time.time()
    for url in urls:
        future=p.submit(get,url)
        l.append(future)
    p.shutdown(wait=True)


    print(\'主\',time.time()-start)

View Code

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import time,random,os
import requests



def get(ur1):
    print(\'%s GET %s\'%(os.getpid(),ur1))
    time.sleep(3)
    response=requests.get(ur1)
    if response.status_code==200:
        res=response.text
    else:
        res=\'下载失败\'
    parse(res)

def parse(res):
    time.sleep(1)
    print(\'%s 解析结果为%s\'%(os.getpid(),len(res)))

if __name__ == \'__main__\':
    urls=[
        \'https://www.baidu.com\'
        \'https://www.youku.com\'
        \'https://www.wangyiyun.com\'
        \'https://www.baidu.com\'
        \'https://www.baidu.com\'
        \'https://www.baidu.com\'

    ]

    p=ProcessPoolExecutor(9)
    l=[]
    start=time.time()
    for url in urls:
        future=p.submit(get,url)
        l.append(future)
    p.shutdown(wait=True)


    print(\'主\',time.time()-start)

View Code

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from threading import current_thread
import time, random, os
import requsets


def get(url):
    print(\'%s GET %s\' % (current_thread().name, url))
    time.sleep(3)
    respose = requsets.get(url)
    if respose.status_code == 200:
        res = respose.text
    else:
        res = \'下载失败\'
    return res


def parse(future):
    time.sleep(1)
    res = future.result()
    print(\'%s 解析结果为%s\' % (current_thread().name, len(res)))


if __name__ == \'__main__\':
    urls = [
        \'https://www.baidu.com\'
        \'https://www.youku.com\'
        \'https://www.wangyiyun.com\'
        \'https://www.baidu.com\'
        \'https://www.baidu.com\'
        \'https://www.baidu.com\'

    ]

    p = ProcessPoolExecutor(9)

    start = time.time()
    for url in urls:
        future = p.submit(get, url)
        # 异步调用：提交完一个任务之后，不是原地等待，
        # 而是直接执行下一行代码，会导致任务是并发执行的，
        # 结果future对象会在任务运行完毕后自动传给回调函数
        future.add_done_callback(parse)  # parse会在任务运行完毕后自动触发，然后接收一个参数future对象

    p.shutdown(wait=True)

    print(\'主\', time.time() - start)
    print(\'主\', os.getpid())

View Code

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from threading import current_thread
import time,random,os
import requests

def get(url):
    print(\'%s GET %s\'%(current_thread().name,url))
    time.sleep(3)
    response=requests.get(url)
    if response.status_code==200:
        res=response.text
    else:
        res=\'下载失败\'
    return res

def parse(future):
    time.sleep(1)
    res=future.result()
    print(\'%s 解析结果为%s\'%(current_thread().name,len(res)))

if __name__ == \'__main__\':
    urls = [
        \'https://www.baidu.com\'
        \'https://www.youku.com\'
        \'https://www.wangyiyun.com\'
        \'https://www.baidu.com\'
        \'https://www.baidu.com\'
        \'https://www.baidu.com\'

    ]

    p=ThreadPoolExecutor(4)

    for url in urls:
        future=p.submit(get,url)
        future.add_done_callback(parse)

    p.shutdown(wait=True)
    print(\'主\',current_thread().name)

View Code

 

二：协程

协程介绍

协程是单线程下的并发，又称微线程，英文名 Coroutine

一句话说明什么是线程：协程是一种后能耗态的轻量级线程，即协程是由用户程序自己控制调度的。

需要强调的是：

#1. python的线程属于内核级别的，即由操作系统控制调度（如单线程遇到io或执行时间过长就会被迫交出cpu执行权限，切换其他线程运行）
#2. 单线程内开启协程，一旦遇到io，就会从应用程序级别（而非操作系统）控制切换，以此来提升效率（！！！非io操作的切换与效率无关）

对比操作系统控制线程的切换，用户在单线程内控制协程的切换

优点如下：

#1. 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
#2. 单线程内就可以实现并发的效果，最大限度地利用cpu

缺点如下：

#1. 协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程
#2. 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程

总结协程特点：

1、必须在只有一个单线程里实现并发

2、修改共享数据不需要加锁

3、用户程序里总结保存多个控制流的上下文栈

4、附加：一个协程遇到IO操作自动切换到其它协程（如何实现检测IO，yield，greenlet都无法实现，就用到了gevent模块（select机制））

 

基于单线程下实现并发，只有一个主线程（如下图：可利用的CPU只有一个）的情况下实现并发，并发的本质：切换+保存状态

CPU正在运行一个任务，会在两种情况下自习其他任务（切换由操作系统强制控制），一种情况是该任务发生了阻塞，另外一种情况是该任务占用时间过长或有一个优先级更高的程序代替了它

#串行执行
import time
def func1():
    for i in range(10000):
        i+1

def func2():
    for i in range(10000):
        i+1

start=time.time()
func1()
func2()
stop=time.time()
print(stop -start)

串行执行

 

#1 yiled可以保存状态，yield的状态保存与操作系统的保存线程状态很像，但是yield是代码级别控制的，更轻量级
#2 send可以把一个函数的结果传给另外一个函数，以此实现单线程内程序之间的切换  

 

#基于yield并发执行
import time
def func1():
    while True:
        print(\'func1\')
        10000+1
        yield

def func2():
    g=func1()
    for i in range(10000):

        print(\'func2\')
        time.sleep(100)
        i+1
        next(g)

start=time.time()
func2()
stop=time.time()
print(stop-start)

yield并发执行

 ps：在介绍进程理论时，提及进程的三种执行状态，而线程才是执行单位，所以可以将上图理解为线程的三种状态