Day37 python基础--并发编程基础6

一，池：concurrent 并发编程中，池的概念

　　concurrent.futures中的进程池 和 线程池：

　　　　1.实例化线程池 ThreadPoolExcutor     一般线程池内线程数为5*cpu核心数

　　　　　实例化进程池 ProcessPoolExcutor   一般进程池内进程数为cpu核心数的1倍或2倍

　　　　2.异步提交任务 submit/map                 对应进程池Pool：apply，apply_async

　　　　3.阻塞直到任务完成 shutdown             对应进程池Pool：close：关闭进程池，不能再提交任务，join：阻塞直到进程池内任务完成 

　　　　4.获取子线程的返回值 result()              对应进程Pool：get()

　　　　5.回调函数 add_done_callback(回调函数 名)   对应进程Pool：callback=回调函数名     

#线程池
import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print('thread',i)
    time.sleep(1)
    print('thread %s end'%i)
tp = ThreadPoolExecutor(5)   #设置线程池中的线程数为5
ret = tp.submit(func,1)          #submit提交任务，按位传参
tp.shutdown()                    #close与jion的结合，关闭任务提交，并等待任务完成后结束阻塞
print('主线程')

#同步的形式提交任务
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
def func(i):
    print('thread',i,currentThread().ident)
    time.sleep(1)
    print('thread %s end'%i)
    return '线程%s已完成'%i

tp = ThreadPoolExecutor(5)
tp_lst = []
for i in range(20):
    ret = tp.submit(func,i)
    print(ret.result())   #阻塞，得到返回值后结束阻塞

#异步形式提交任务
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
def func(i):
    print('thread',i,currentThread().ident)
    time.sleep(1)
    print('thread %s end'%i)
    return '线程%s已完成'%i
tp = ThreadPoolExecutor(5)
tp_lst = []
for i in range(20):
    ret = tp.submit(func,i)
    tp_lst.append(ret)
#tp.shutdown()  #取返回值时自带阻塞，可以不等全部执行完再取值
for tp in tp_lst:
    print(tp.result())
print('主线程')

#使用map提交任务
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread
def func(i):
    print('thread',i,currentThread().ident)
    time.sleep(1)
    print('thread %s end'%i)
    return '线程%s已完成'%i
tp = ThreadPoolExecutor(5)
ret = tp.map(func,range(20))   #传入任务，和任务数量
for i in ret:print(i)    # 直接取得返回值

#线程池的回调函数
import time
from concurrent.futures import ThreadPoolExecutor
from threading import currentThread   
def func(i):
    print('thread',i,currentThread().ident)   #查看线程id
    time.sleep(1)
    print('thread %s end'%i)
    return '线程%s已完成'%i

#线程池的回调函数，是由子线程完成的
def call_back(arg):
    print('call_back',currentThread().ident)
    print(arg.result())

tp = ThreadPoolExecutor(5)
tp_lst = []
for i in range(20):
    ret = tp.submit(func,i).add_done_callback(call_back)   #提交回调函数
print('主线程',currentThread().ident)




#进程池的回调函数
import os
import time
from concurrent.futures import ProcessPoolExecutor
from threading import currentThread
def func(i):
    print('thread',i,os.getpid())
    time.sleep(1)
    print('thread %s end'%i)
    return '进程%s已完成'%i

#进程池的回调函数，是由主进程程完成的
def call_back(arg):
    print('call_back',os.getpid())
    print(arg.result())

if __name__ == '__main__':
    tp = ProcessPoolExecutor(5)
    tp_lst = []
    for i in range(20):
        ret = tp.submit(func,i).add_done_callback(call_back)
    print('主进程',currentThread().ident)

二，协程：又称纤程

　　线程的本质：一条线程在多个任务之间来回切换

　　切换这个动作是浪费时间的

　　对于cpu和操作系统来说，协程是不存在的，它们只能看到线程

#协程的概念就像线程中的代码交替执行，但从表面上来看它们是并发执行
#yield实现生产者消费者模型
def consumer():
    g = producer()
    for num in g:
        print(num)

def producer():
    for i in range(1000):
        yield i

consumer()

#更好的利用协程

1.一个线程的执行明确的切分开
2.协程中的两个任务：协程会帮助你记住哪个任务执行到哪个位置上了，并且实现安全的切换
3.一个任务不得不陷入阻塞了，在这个任务阻塞的过程中切换到另一个任务继续执行
4.协程中只要还有任务需要执行，你的当前线程永远不会阻塞
5.利用协程在多个任务陷入阻塞的时候进行切换来保证一个线程在处理多个任务的时候总是处于忙碌状态，使操作系统能多分配时间片，能够更加充分的利用cpu，抢占更多的时间片
6.无论是进程还是线程都是由操作系统来切换的，开启过多的线程或进程会给操作系统的调度造成负担
7.如果我们是使用协程，协程在程序之间的切换操作系统感知不到，无论开启多少个协程操作系统的调度不会有任何压力

#数据安全方面：
   协程的本质就是一条线程，以代码为单位进行切换，所以完全不会产生数据安全的问题

　　

　　协程模块：

　　　　greenlet：gevent底层，协程，切换的模块

　　　　gevent：可以直接使用，gevent能提供更全面的功能

#greenlet只能完成基本的协程切换
import time
from greenlet import greenlet
def eat():
    print('eating 1')
    g2.switch()     #切换至协程g2
    time.sleep(1)
    print('eating 2')

    g2.switch()
def play():
    print('playing 1')
    g1.switch()      #切换至协程g1
    time.sleep(1)
    print('playing 2')
    g1.switch()

g1 = greenlet(eat)
g2 = greenlet(play)
g1.switch()  #在主线程，协程g1,协程g2中切换

#gevent能够自动的检测阻塞事件，遇见阻塞就会进行切换，但是能识别的阻塞有限
import time
import gevent

def eat():
    print('eating 1')
    gevent.sleep(1)   #time.sleep(1)这个阻塞，gevent无法识别
    print('eating 2')

def play():
    print('playing 1')
    gevent.sleep(1)  # time.sleep(1)
    print('playing 2')

g1 = gevent.spawn(eat)    #自动的检测阻塞事件，遇见阻塞就会进行切换
g2 = gevent.spawn(play)
g1.join()     #阻塞直到g1结束
g2.join()      #阻塞直到g2结束

#monkey模块的patch_all方法，导入其他模块中的内部带有的一些阻塞事件
from gevent import monkey
monkey.patch_all()  #打包下面import的模块的所有阻塞，然后gevent就能识别这些阻塞了
import time
import gevent

def eat():
    print('eating 1')
    time.sleep(1)    #gevent可以识别这些阻塞了
    print('eating 2')

def play():
    print('playing 1')
    time.sleep(1)
    print('playing 2')

g1 = gevent.spawn(eat)    #自动的检测阻塞事件，遇见阻塞就会进行切换
g2 = gevent.spawn(play)
#g1.join()  #阻塞直到g1结束
#g2.join()  #阻塞直到g2结束
gevent.joinall([g1,g2])    #直接传入一个协程对象的列表，可以直接阻塞全部协程

#value取得返回值
from gevent import monkey;monkey.patch_all()
import time
import gevent

def eat():
    print('eating 1')
    time.sleep(1)
    print('eating 2')
    return 'eating finished'

def play():
    print('playing 1')
    time.sleep(1)
    print('playing 2')
    return 'playing finished'

g1 = gevent.spawn(eat)    #自动的检测阻塞事件，遇见阻塞就会进行切换
g2 = gevent.spawn(play)
gevent.joinall([g1,g2])  #阻塞,直到g1,g2协程任务结束
print(g1.value)
print(g2.value)

#协程总结

from gevent import monkey;monkey.patch_all() :
#导入其他模块内部带有一些的阻塞事件

g1 = gevent.spawn(函数名，函数的参数) :
#产生了一个协程任务，在遇到IO操作的时候帮助我们在多任务之间切换，有些阻塞它识别不了,spawn()按位传入参数

join():
#阻塞,直到某个人物被执行完毕

join_all()：
#传入一个列表，阻塞列表内的所有协程任务

value属性：
获取返回值

#协程实例：爬虫
#任务越多，协程所提高的效率越明显

from gevent import monkey;monkey.patch_all()
import gevent
import time
import requests
url_lst = [
    'http://www.baidu.com',
    'http://www.4399.com',
    'http://www.sogou.com',
    'http://www.sohu.com',
    'http://www.sina.com',
    'http://www.jd.com',
    'http://www.7k7k.com',
    'https://www.luffycity.com/home',
    'https://www.douban.com',
]

def get_url(url):
    response = requests.get(url)
    if response.status_code == 200:  #状态码 200 正常的结果
        print(url,len(response.text))  #网页源代码
start = time.time()
g_lst = []
for url in url_lst:
    g = gevent.spawn(get_url,url)    # get_url(url)  #2.1261658668518066
    g_lst.append(g)
gevent.joinall(g_lst)
print(time.time()-start)    #1.0275120735168457

#协程实例2：socket模块+协程，实现并发通信

#server端：
from gevent import monkey;monkey.patch_all()
import socket
import gevent
from threading import currentThread
def talk(conn):
    print('->',currentThread())
    while 1:
        conn.send(b'hello')
        conn.recv(1024)

sk = socket.socket()
sk.bind(('127.0.0.1',9000))
sk.listen()

while True:
    conn,addr = sk.accept()   #主线程 accept 基本上都是阻塞
    gevent.spawn(talk,conn)   #在协程内可以实现并发通信



#client端
import socket
from threading import Thread
def client():
    sk = socket.socket()
    sk.connect(('127.0.0.1',9000))
    while 1:
        print(sk.recv(1024))
        sk.send(b'byebye')
for i in range(5):
    Thread(target=client).start()