concurrent.futures进线程池和协程

时间：2024-11-26 12:38:02

concurrent.futures

异步执行进程线程池的模块，一个抽象类，定义submit，map，shutdown方法

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor

import time,os,random

def task(n):

    print(os.getpid(),'is running')

    time.sleep(random.randint(,))

    return n**

if __name__ == '__main__':

    p = ProcessPoolExecutor()

    obj = p.map(task,range())

    p.shutdown()

    print(list(obj))

进程池

线程池就是

ProcessPoolExecutor换成
ThreadPoolExecutor
import os,time,requests,re

from concurrent.futures import ThreadPoolExecutor

def get_page(url):

    print(url)

    ret = requests.get(url).text

    return {'url':url,'text':ret}

def get_p(res):

    res  = res.result()

    rep = re.compile(r'<a href="/films/\d+" title="(?P<name>.*?)" class="image-link.*?<p class="star">(?P<star>.*?)</p>',re.S)

    ret = rep.finditer(res['text'])

    with open('db.txt','a',encoding='utf-8')as f:

        for i in ret:

            s = "name:%s"%i.group("name")+i.group('star')

            f.write(s+'\n')

if __name__ == '__main__':

    t = time.time()

    p = ThreadPoolExecutor()

    urls = [

        'http://maoyan.com/board/7',

        'http://maoyan.com/board/6',

        'http://maoyan.com/board/1',

        'http://maoyan.com/board/2',

        'http://maoyan.com/board/4',

    ]

    for url in urls:

        p.submit(get_page,url).add_done_callback(get_p)

    p.shutdown()

    print(time.time()-t)

回调函数

协程

协程：是单线程下的并发，又称微线程，纤程。英文名Coroutine。一句话说明什么是线程：协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。、



需要强调的是：






#1. python的线程属于内核级别的，即由操作系统控制调度（如单线程遇到io或执行时间过长就会*交出cpu执行权限，切换其他线程运行）

#2. 单线程内开启协程，一旦遇到io，就会从应用程序级别（而非操作系统）控制切换，以此来提升效率（！！！非io操作的切换与效率无关）






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




优点如下：






#1. 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级

#2. 单线程内就可以实现并发的效果，最大限度地利用cpu






缺点如下：






#1. 协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程

#2. 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程






总结协程特点：




    

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

  2. 修改共享数据不需加锁
    3. 

  3. 用户程序里自己保存多个控制流的上下文栈
    4. 

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



 


Greenlet






from greenlet import greenlet

def eat(name):

    print('%s eat 1' %name)

    g2.switch('egon')

    print('%s eat 2' %name)

    g2.switch()

def play(name):

    print('%s play 1' %name)

    g1.switch()

    print('%s play 2' %name)

g1=greenlet(eat)

g2=greenlet(play)

g1.switch('egon')#可以在第一次switch时传入参数，以后都不需要




单纯切换，io阻塞无用




Gevent介绍


Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度。


记得打猴子补丁






from gevent import monkey;monkey.patch_all()

import gevent

import time

def eat():

    print('eat food 1')

    time.sleep()

    print('eat food 2')

def play():

    print('play 1')

    time.sleep()

    print('play 2')

g1=gevent.spawn(eat)

g2=gevent.spawn(play_phone)

gevent.joinall([g1,g2])

print('主')










from gevent import monkey;monkey.patch_all()

import gevent

import requests

import time

def get_page(url):

    print('GET: %s' %url)

    response=requests.get(url)

    if response.status_code == :

        print('%d bytes received from %s' %(len(response.text),url))

start_time=time.time()

gevent.joinall([

    gevent.spawn(get_page,'https://www.python.org/'),

    gevent.spawn(get_page,'https://www.yahoo.com/'),

    gevent.spawn(get_page,'https://github.com/'),

])

stop_time=time.time()

print('run time is %s' %(stop_time-start_time))




协程爬虫








from gevent import monkey;monkey.patch_all()

from socket import *

import gevent

#如果不想用money.patch_all()打补丁,可以用gevent自带的socket

# from gevent import socket

# s=socket.socket()

def server(server_ip,port):

    s=socket(AF_INET,SOCK_STREAM)

    s.setsockopt(SOL_SOCKET,SO_REUSEADDR,)

    s.bind((server_ip,port))

    s.listen()

    while True:

        conn,addr=s.accept()

        gevent.spawn(talk,conn,addr)

def talk(conn,addr):

    try:

        while True:

            res=conn.recv()

            print('client %s:%s msg: %s' %(addr[],addr[],res))

            conn.send(res.upper())

    except Exception as e:

        print(e)

    finally:

        conn.close()

if __name__ == '__main__':

    server('127.0.0.1',)




协程服务端








from threading import Thread

from socket import *

import threading

def client(server_ip,port):

    c=socket(AF_INET,SOCK_STREAM) #套接字对象一定要加到函数内，即局部名称空间内，放在函数外则被所有线程共享，则大家公用一个套接字对象，那么客户端端口永远一样了

    c.connect((server_ip,port))

    count=

    while True:

        c.send(('%s say hello %s' %(threading.current_thread().getName(),count)).encode('utf-8'))

        msg=c.recv()

        print(msg.decode('utf-8'))

        count+=

if __name__ == '__main__':

    for i in range():

        t=Thread(target=client,args=('127.0.0.1',))

        t.start()




多线程并发多个客户端




 多线程+协程！！
关键：加入猴子补丁monkey后，所有子线程遇到IO会阻塞，所有程序卡住不运行
解决方法：设置


from gevent import monkey;monkey.patch_all(thread=False)

https://*.com/questions/9192539/using-gevent-monkey-patching-with-threading-makes-thread-work-serially




相关文章