码迷,mamicode.com
首页 > 编程语言 > 详细

python-进程池与线程池,协程

时间:2018-09-11 01:09:36      阅读:321      评论:0      收藏:0      [点我收藏+]

标签:不能   size   yield   listen   shutdown   www.   sleep   cti   最大   

一、进程池与线程池

实现并发的手段有两种,多线程和多进程。注:并发是指多个任务看起来是同时运行的。主要是切换+保存状态。

当我们需要执行的并发任务大于cpu的核数时,我们需要知道一个操作系统不能无限的开启进程和线程,通常有几个核就开几个进程,如果进程开启过多,就无法充分利用cpu多核的优势,效率反而会下降。这个时候就引入了进程池线程池的概念。

池的功能就是限制启动的进程数或线程数

concurent.future模块:

concurrent.futures模块提供了高度封装的异步调用接口

ProcessPoolExecutor: 进程池,提供异步调用

p = ProcessPoolExecutor(max_works)对于进程池如果不写max_works:默认的是cpu的数目,默认是4个

ThreadPoolExecutor:线程池,提供异步调用   
p = ThreadPoolExecutor(max_works)对于线程池如果不写max_works:默认的是cpu的数目*5

 

补充:

提交任务的两种方式:
# 同步调用:提交完一个任务之后,就在原地等待,等待任务完完整整地运行完毕拿到结果后,再执行下一行代码,会导致任务是串行执行的
# 异步调用:提交完一个任务之后,不在原地等待,结果???,而是直接执行下一行代码,会导致任务是并发执行的

进程池从无到有创建进程后,然后会固定使用进程池里创建好的进程去执行所有任务,不会开启其他进程

# 基本方法
#submit(fn, *args, **kwargs)
异步提交任务

#map(func, *iterables, timeout=None, chunksize=1) 
取代for循环submit的操作

#shutdown(wait=True) 
相当于进程池的pool.close()+pool.join()操作
wait=True,等待池内所有任务执行完毕回收完资源后才继续
wait=False,立即返回,并不会等待池内的任务执行完毕
但不管wait参数为何值,整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前

#result(timeout=None)
取得结果

#add_done_callback(fn)
回调函数
技术分享图片
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=下载失败
    return res

def parse(future):
    time.sleep(1)
    res=future.result()
    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)

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

    p.shutdown(wait=True)


    print(,time.time()-start)
    print(,os.getpid())
test

线程池与进程池相比 他们的同步执行和异步执行是一样的:

技术分享图片
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.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=ThreadPoolExecutor(4)
    
    for url in urls:
        future=p.submit(get,url)
        future.add_done_callback(parse)

    p.shutdown(wait=True)

    print(,current_thread().name)
test

map函数:

技术分享图片
# 我们的那个p.submit(task,i)和map函数的原理类似。我们就
# 可以用map函数去代替。更减缩了代码
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import os, time, random


def task(n):
    print([%s] is running % os.getpid())
    time.sleep(random.randint(1, 3))  # I/O密集型的,,一般用线程,用了进程耗时长
    return n ** 2


if __name__ == __main__:
    p = ProcessPoolExecutor()
    obj = p.map(task, range(10))
    p.shutdown()  # 相当于close和join方法
    print(= * 30)
    print(obj)  # 返回的是一个迭代器
    print(list(obj))
View Code

回调函数(知乎):https://www.zhihu.com/question/19801131/answer/27459821

二、协程

在单线程的情况下实现并发。

遇到IO就切换就可以降低单线程的IO时间,从而最大限度地提升单线程的效率。

实现并发是让多个任务看起来同时运行(切换+保存状态),cpu在运行一个任务的时候,会在两种情况下去执行其他的任务,一种是遇到了I/O操作,一种是计算时间过长。其中第二种情况使用线程并发并不能提升效率,运算密集型的并发反而会降低效率。

技术分享图片
#串行执行
import time

def func1():
    for i in range(10000000):
        i+1

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

start = time.time()
func1()
func2()
stop = time.time()
print(stop - start)#1.675490379333496
串行执行
技术分享图片
#基于yield并发执行
import time
def func1():
    while True:
        print(func1)
        100000+1
        yield

def func2():
    g=func1()
    for i in range(10000000):
        print(func2)
        time.sleep(100)
        i+1
        next(g)

start=time.time()
func2()
stop=time.time()
print(stop-start)
基于yield并发执行

yield复习:

函数中只有有yield,这个函数就变成了一个生成器,调用函数不会执行函数体代码,会得到一个返回值,返回值就是生成器对象。

技术分享图片
def yield_test(n):
    for i in range(n):
        yield call(i)
        print("i=",i)
    #做一些其它的事情
    print("do something.")
    print("end.")

def call(i):
    return i*2

#使用for循环
for i in yield_test(5):
    print(i,",")
test

协程的本质就是在单线程下,由用户自己控制一个任务遇到IO操作就切换到另一个任务去执行,以此来提升效率。

Gevent:

gevent是第三方库,通过greenlet实现协程,其基本思想是:

当一个greenlet遇到IO操作时,比如访问网络,就自动切换到其他的greenlet,等到IO操作完成,再在适当的时候切换回来继续执行。由于IO操作非常耗时,经常使程序处于等待状态,有了gevent为我们自动切换协程,就保证总有greenlet在运行,而不是等待IO。

由于切换是在IO操作时自动完成,所以gevent需要修改Python自带的一些标准库,这一过程在启动时通过monkey patch完成:

我们用等待的时间模拟IO阻塞 在gevent模块里面要用gevent.sleep(5)表示等待的时间 要是我们想用time.sleep(),就要在最上面导入from gevent import monkey;monkey.patch_all()这句话 如果不导入直接用time.sleep(),就实现不了单线程并发的效果了

注:猴子补丁需要在第一行就运行

技术分享图片
from gevent import monkey;monkey.patch_all()
from gevent import spawn,joinall #pip3 install gevent
import time

def play(name):
    print(%s play 1 %name)
    time.sleep(5)
    print(%s play 2 %name)

def eat(name):
    print(%s eat 1 %name)
    time.sleep(3)
    print(%s eat 2 %name)


start=time.time()
g1=spawn(play,lxx)
g2=spawn(eat,lxx)

# g1.join()
# g2.join()
joinall([g1,g2])
print(,time.time()-start)
test

gevent.spawn()”方法会创建一个新的greenlet协程对象,并运行它。”gevent.joinall()”方法会等待所有传入的greenlet协程运行结束后再退出,这个方法可以接受一个”timeout”参数来设置超时时间,单位是秒。

在单线程内实现socket并发:

技术分享图片
from gevent import monkey;monkey.patch_all()
from socket import *
from gevent import spawn

def comunicate(conn):
    while True:  # 通信循环
        try:
            data = conn.recv(1024)
            if len(data) == 0: break
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()


def server(ip, port, backlog=5):
    server = socket(AF_INET, SOCK_STREAM)
    server.bind((ip, port))
    server.listen(backlog)

    while True:  # 链接循环
        conn, client_addr = server.accept()
        print(client_addr)

        # 通信
        spawn(comunicate,conn)

if __name__ == __main__:
    g1=spawn(server,127.0.0.1,8080)
    g1.join()
server
技术分享图片
from threading import Thread,current_thread
from socket import *

def client():
    client=socket(AF_INET,SOCK_STREAM)
    client.connect((127.0.0.1,8080))

    n=0
    while True:
        msg=%s say hello %s %(current_thread().name,n)
        n+=1
        client.send(msg.encode(utf-8))
        data=client.recv(1024)
        print(data.decode(utf-8))

if __name__ == __main__:
    for i in range(500):
        t=Thread(target=client)
        t.start()
client

 

python-进程池与线程池,协程

标签:不能   size   yield   listen   shutdown   www.   sleep   cti   最大   

原文地址:https://www.cnblogs.com/mangM/p/9623585.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!