线程池、进程池(concurrent.futures模块)和协程

一、线程池
1、concurrent.futures模块
介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
在这个模块中进程池和线程池的使用方法完全一样


这里就只介绍ThreadPoolExecutor的使用方法，顺便对比multiprocessing的Pool进程池

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

map(func, *iterables, timeout=None, chunksize=1) ：取代for循环submit的操作，iterables的每个元素都作为参数传给func

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

result(timeout=None)：取得返回值

add_done_callback(fn)：设置回调函数

3、例子：(对比multiprocessing的Pool模块)
3-1、
import time
from concurrent.futures import ThreadPoolExecutor

def func(i):
    print('thread',i)
    time.sleep(1)
    print('thread %s end'%i)

tp = ThreadPoolExecutor(5)  # 相当于tp = Pool(5)
tp.submit(func,1)           # 相当于tp.apply_async(func,args=(1,))
tp.shutdown()               # 相当于tp.close()  +  tp.join()
print('主线程')


3-2、
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)

tp = ThreadPoolExecutor(5)
for i in range(20):
    tp.submit(func,i) 
tp.shutdown()  # shutdown一次就够了，会自动把所有的线程都join()
print('主线程')



3-3、返回值
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 i * '*'

tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
    ret = tp.submit(func,i)
    ret_lst.append(ret)

for ret in ret_lst:
    print(ret.result())  # 相当于ret.get()

print('主线程')


3-4、map
map接收一个函数和一个可迭代对象
可迭代对象的每一个值就是函数接收的实参，可迭代对象的长度就是创建的线程数量
map可以直接拿到返回值的可迭代对象(列表)，循环就可以获取返回值

import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print('thread',i)
    time.sleep(1)
    print('thread %s end'%i)
    return i * '*'

tp = ThreadPoolExecutor(5)
ret = tp.map(func,range(20))
for i in ret:
    print(i)




3-5、回调函数
    回调函数在进程池是由主进程实现的
    回调函数在线程池是由子线程实现的


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 i * '*'

def call_back(arg):
    print('call back : ',currentThread().ident)
    print('ret : ',arg.result())  # multiprocessing的Pool回调函数中的参数不需要get(),这里需要result()

tp = ThreadPoolExecutor(5)
ret_lst = []
for i in range(20):
    tp.submit(func,i).add_done_callback(call_back)  # 使用add_done_callback()方法实现回调函数 
print('主线程',currentThread().ident)


从结果可以看出：
    子线程func执行完毕之后才去执行callback回调函数
    子线程func的返回值会作为回调函数的参数

4、总结
线程池
    实例化线程池  ThreadPoolExcutor    5*cpu_count
    异步提交任务  submit / map
    阻塞直到任务完成 shutdown
    获取子线程的返回值 result
    回调函数      add_done_callback
    



二、协程概念介绍
1、介绍
进程 ：计算机中最小的资源分配单位
线程 ：计算机中能被cpu执行的最小单位
协程(纤程)：一条线程在多个任务之间来回切换就叫协程
          切换这个动作是浪费时间的
          对于CPU、操作系统来说协程是不存在的
          他们只能看到线程

协程的本质就是一条线程在多个任务之间来回切换，所以完全不会产生数据安全的问题

协程的理解：
把一个线程的执行明确的切分开
比如有两个任务，使用协程它帮助你记住哪个任务执行到哪个位置上了，并且实现安全的切换
当一个任务陷入阻塞，在这个任务阻塞的过程中切换到另一个任务中执行另一个任务
你的程序只要还有任务需要执行 你的当前线程永远不会阻塞
利用协程在多个任务陷入阻塞的时候进行切换来保证一个线程在处理多个任务的时候总是忙碌的
能够更加充分的利用CPU，抢占更多的时间片
无论是进程、还是线程都是由操作系统来切换的，开启过多的线程、进程会给操作系统的调度增加负担
如果我们是使用协程，协程在程序之间的切换操作系统感知不到，无论开启多少个协程对操作系统来说总是一个线程
操作系统的调度不会有任何压力


2、用生活中的例子来解释：
协程：
比如你自己一个人
做饭(协程)需要半个小时，你可以先洗米，然后把米饭交给电饭煲煮，这个时候煮饭就陷入阻塞了，你就可以去做其他家务了，
比如这个时候你可以去洗衣服(协程)，你把衣服放进洗衣机后，这个任务也陷入阻塞了，然后你又可以去做其他事情，
比如这个时候你可以收拾屋子 (协程)，然后在你收拾屋子的时候，米饭煮好了，你就去关电饭煲，过一段时间，衣服也洗好了，
你就去关洗衣机，晾衣服，其他时间你都在收拾屋子，那么这样你的时间就利用得很充分了。


多线程(多进程)：
上面的任务，你请了几个人帮你一起做，每个人都只做自己的那件事，
比如你做饭，然后一个人洗衣服，另一个人收拾屋子，这样的话在阻塞的时间里，每个人都是在等待的状态，没有充分利用时间，
且成本高(你请人需要人工)


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



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

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


5、总结：
必须在只有一个单线程里实现并发
修改共享数据不需加锁
用户程序里自己保存多个控制流的上下文栈
线程的调度是操作系统级别的
协程的调度是用户级别的

三、greenlet模块和gevent模块(都是扩展模块)
1、介绍
greenlet：是gevent的底层，协程切换的模块，但是当切到一个任务执行时如果遇到io，那就原地阻塞，仍然是没有解决遇到IO自动切换来提升效率的问题。
gevent：基于greenlet优化而来的一个模块，gevent能提供更全面的功能，遇到io操作会自动切换任务(所以一般直接用这个模块就好)。


gevent用法介绍：
g1=gevent.spawn(func,1,,2,3,x=4,y=5)创建一个协程对象g1，spawn括号内第一个参数是函数名，后面可以有多个参数，可以是位置实参或关键字实参，都是传给你指定的函数

g2=gevent.spawn(func2)

g1.join()   #等待g1结束

g2.join()   #等待g2结束

上面两个join可以写成一个：gevent.joinall([g1,g2])   # 参数是列表类型 

g1.value   #拿到func的返回值


注意：在gevent模块中，有些阻塞它是不认识的，比如time模块的sleep，如果直接导入time模块，使用time.sleep()，在gevent模块的协程中并不会阻塞，因为gevent不认识time模块，
那么如何解决呢？
在导入time模块前先写上下面这两行代码，再导入time模块，这样time模块的内容它就认识了，time.sleep()也会阻塞了：
from gevent import monkey
monkey.patch_all()  # patch_all就是把下面的模块的阻塞打成一个包，认识他们
import time

而在greenlet模块中，time的模块的阻塞它本身就是认识的。


2、例子
2-1、greenlet例子：
import time
from greenlet import greenlet
def cooking():
    print('cooking 1')
    g2.switch()      # 切换到g2，让g2的函数工作
    time.sleep(1)
    print('cooking 2')

def watch():
    print('watch TV 1')
    time.sleep(1)
    print('watch TV 2')
    g1.switch()    # 切换到g1，让g1的函数工作

g1 = greenlet(cooking)
g2 = greenlet(watch)
g1.switch()        # 切换到g1，让g1的函数工作

greenlet的缺陷：很显然greenlet实现了协程的切换功能，可以自己设置什么时候切，在哪切，但是它遇到阻塞并没有自动切换，
因此并不能提高效率。所以一般我们都使用gevent模块实现协程



2-2、gevent例子：
from gevent import monkey
monkey.patch_all()
import time
import gevent

def cooking():
    print('cooking 1')
    time.sleep(1)
    print('cooking 2')

def watch():
    print('watch TV 1')
    time.sleep(1)
    print('watch TV 2')

g1 = gevent.spawn(cooking) # 自动检测阻塞事件，遇见阻塞了就会进行切换
g2 = gevent.spawn(watch)

g1.join()    # 阻塞直到g1结束
g2.join()    # 阻塞直到g2结束



2-3、gevent例子2：

import gevent
def cooking(i):
    print('%s号在煮饭' %i)
    return i

g_lst = []
for i in range(10):
    g = gevent.spawn(cooking,i) # 函数名，参数
    g_lst.append(g)   # 把协程对象放入列表

for g in g_lst:
    g.join()
    print(g.value)  # 打印返回值

# gevent.joinall(g_lst) # joinall一次性把全部对象都阻塞

2-4、协程名：
from gevent import monkey
monkey.patch_all()
import time
import gevent
from threading import currentThread

def cooking():
    print('cooking name:',currentThread().getName())
    print('cooking 1')
    time.sleep(1)
    print('cooking 2')

def watch():
    print('watch name:', currentThread().getName())
    print('watch TV 1')
    time.sleep(1)
    print('watch TV 2')

g1 = gevent.spawn(cooking) 
g2 = gevent.spawn(watch)

g1.join()  
g2.join()  

# gevent.joinall([g1,g2])

结果：
cooking name: DummyThread-1    #  Dummy的意思是假的，即协程是假线程，它只是同一个线程在任务间来回切换
cooking 1
watch name: DummyThread-2
watch TV 1
cooking 2
watch TV 2




2-4、基于协程的爬虫例子：
from gevent import monkey
monkey.patch_all()
import time
import gevent
import requests  # 扩展模块

url_lst = [
'http://www.baidu.com',
'http://www.4399.com',
'http://www.sohu.com',
'http://www.jd.com',
'http://www.sina.com',
'https://www.douban.com',
'http://www.sohu.com',
'http://www.baidu.com',
'http://www.4399.com',
'http://www.sohu.com',
'http://www.jd.com',
'http://www.sina.com',
'https://www.douban.com',
'http://www.sohu.com',
'http://www.baidu.com',
'http://www.4399.com',
'http://www.sohu.com',
'http://www.jd.com',
'http://www.sina.com',
'https://www.douban.com',
'http://www.sohu.com',
'http://www.baidu.com',
'http://www.4399.com',
'http://www.sohu.com',
'http://www.jd.com',
'http://www.sina.com',
'https://www.douban.com',
'http://www.sohu.com'
]

def get_url(url):
    response = requests.get(url)
    if response.status_code == 200:  # response.status_code的值是200的时候才代表爬取成功
        print(url,len(response.text)) # response.text是爬取的网页内容，这里只打印一下内容的长度

# 普通方式爬取网页
# start = time.time()
# for url in url_lst:
#     get_url(url)
# print(time.time()-start)    # 使用时间：5.616770267486572


# 使用协程爬取网页
start = time.time()
g_lst = []
for url in url_lst:
    g = gevent.spawn(get_url,url)
    g_lst.append(g)
gevent.joinall(g_lst)
print(time.time()-start)     # 使用时间：1.8181169033050537




2-5、基于gevent的socket
注意：from gevent import monkey;monkey.patch_all()一定要放到导入socket模块之前，否则gevent无法识别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')
        print(conn.recv(1024))  # 接收的时候阻塞，切换到另一个任务

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

while 1:
    conn,addr = sk.accept()
    gevent.spawn(talk,conn)
    # 这里不需要join了，因为accept会阻塞，而且如果有join了之后，
    # 第一个协程没有运行完毕这里的循环就不会继续走了



Client端：
import socket
from threading import Thread
def client():
    sk = socket.socket()
    sk.connect(('127.0.0.1',8000))
    while 1:
        print(sk.recv(1024))
        sk.send(b'hi')
for i in range(500):  # 开启500个线程，即500个用户
    Thread(target=client).start()