[<_MainThread(MainThread, started 20008)>, ]

2

MainThread, started 20008 表示主线程，Thread-1, started 20232 表示子线程。它会打印出 2 个。

所以 activeCount 的结果为 2

二、同步锁

当多线程争夺锁时，允许第一个获得锁的线程进入临街区，并执行代码。所有之后到达的线程将被阻塞，直到第一个线程执行结束，退出临街区，并释放锁。

多个线程抢占资源的情况：

import time
from threading import Thread
def func():
    global n
    temp = n
    time.sleep(1)
    n = temp -1
n = 100
t_lst = []
for i in range(100):
    t = Thread(target=func)
    t.start()
    t_lst.append(t)
for t in t_lst:t.join()
print(n)

执行输出：99

为啥呢？

明明减了 100 次，结果应该是 0 的。

为啥是 99 呢？难度是 GIL 的问题，但 GIL 是计算 CPU 那一刻的锁

下面开始具体分析：

第一步，每个线程执行 global n:temp =n 此时，temp 等于 100

第二步，当线程设计到 CPU 计算时，向 CPU 发送请求。但是收到 GIL 的限制

同一时刻，只能有一个线程计算。

CPU 计算结果后，返回给线程。线程赋值，并修改全局变量 n

此时 n=99，线程结束

那么其他线程，也是做同样的操作

每个线程赋值 n 等于 99。不管它已经是 99 了。

上面的现象，出现了数据不安全的情况

最后赋值了 100 次，都是 n=99。所以最终结果是 99

怎么解决呢？加锁

from threading import Thread,Lock
def func(lock):
    global n
    lock.acquire()  # 加锁
    temp = n
    n = temp -1
    lock.release()  # 解锁
n = 100
t_lst = []
lock = Lock()  # 创建锁
for i in range(100):
    t = Thread(target=func,args=(lock,))
    t.start()
    t_lst.append(t)
for t in t_lst:t.join()  # 等待所有子线程结束
print(n)

执行输出：0

如果把计算和赋值拆开，就会出现数据不安全的情况

下面写法，不用加锁，也可以得到 0

from threading import Thread
def func():
    global n
    n -= 1
n = 100
for i in range(100):
    t = Thread(target=func)
    t.start()<br>print(n)

执行输出：0

因为默认有一个 GIL 锁，所以每个线程都减等 1。所以最终结果为 0

三、死锁与递归锁

死锁

进程也有死锁与递归锁，在进程那里忘记说了，放到这里一切说了额

所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁

from threading import Lock
lock = Lock()  # 在同一个线程中,能够被一个锁的多个acquire阻住,这种锁就叫做互斥锁
lock.acquire()
lock.acquire()
lock.acquire()

死锁，也叫互斥锁

科学家吃面的问题

要完成一件事情 需要两个必要因素

要想吃到面，需要： 叉子，面

资源的互相抢占的问题 —— 死锁

四个人围着一张桌子，桌子上放着一碗面，碗里有一个叉子

必须拿到叉子，才能吃面。

每个人每次只能吃一口面，吃完就放回去，让下一个人吃。

import time
from threading import Thread,Lock
def eat1(noodle_lock,fork_lock,name):  # 吃面
    noodle_lock.acquire()  # 面条加锁
    print('%s 抢到了面条'%name)
    fork_lock.acquire()  # 叉子加锁
    print('%s 抢到了叉子'%name)
    print('%s 正在吃面'%name)
    fork_lock.release()  # 叉子解锁
    print('%s归还了叉子'%name)
    noodle_lock.release()  # 面条解锁
    print('%s归还了面条' % name)
def eat2(noodle_lock,fork_lock,name):  #也是吃面，不同的是：先抢叉子，再抢面
    fork_lock.acquire()
    print('%s抢到了叉子' % name)
    time.sleep(0.5)
    noodle_lock.acquire()
    print('%s抢到了面'%name)
    print('%s正在吃面'%name)
    noodle_lock.release()
    print('%s归还了面' % name)
    fork_lock.release()
    print('%s归还了叉子' % name)
noodle_lock = Lock()  # 面条锁
fork_lock = Lock()  # 叉子锁
t1 = Thread(target=eat1,args=(noodle_lock,fork_lock,'nazha')).start()
t2 = Thread(target=eat2,args=(noodle_lock,fork_lock,'egon')).start()
t3 = Thread(target=eat1,args=(noodle_lock,fork_lock,'yuan')).start()
t4 = Thread(target=eat2,args=(noodle_lock,fork_lock,'alex')).start()

执行输出：

nazha 抢到了面条

nazha 抢到了叉子

nazha 正在吃面

nazha 归还了叉子

nazha 归还了面条

egon 抢到了叉子

yuan 抢到了面条

发现程序卡住了，为啥呢？

egon 和 yuan 分别抢到了叉子和面条，但是谁也不愿意给对方。互掐在…

那么就出现了死锁现象。

只有一个锁，不会出现死锁。在多个锁的情况下，就会出现死锁

如何解决这个问题呢？使用递归锁

递归锁

在 Python 中为了支持在同一线程中多次请求同一资源，python 提供了可重入锁 RLock。

这个 RLock 内部维护着一个 Lock 和一个 counter 变量，counter 记录了 acquire 的次数，从而使得资源可以被多次 require。直到一个线程所有的 acquire 都被 release，其他的线程才能获得资源

简单例子：

from threading import Thread,RLock
rlock = RLock()  # 创建递归锁
rlock.acquire()  # 加第一个锁
print('***')
rlock.acquire()  # 加第二个锁
print('***')

执行输出：

发现瞬间就输出了，程序没有卡顿，666

结论：

递归锁在同一个线程中对同一个锁多次 acquire 不会产生阻塞

看下面的例子

from threading import Thread,RLock
def func(rlock,flag):
    rlock.acquire()  # 第一道锁
    print(flag*10)
    rlock.acquire()  # 第二道锁
    print(flag * 10)
    rlock.acquire()  # 第三道锁
    print(flag * 10)
    rlock.acquire()  # 第四道锁
    print(flag * 10)
rlock = RLock()  # 创建递归锁
Thread(target=func,args=(rlock,'*')).start()  # 传入递归锁和*
Thread(target=func,args=(rlock,'-')).start()

执行输出：

发现程序卡住了，————-没有输出

为什么？递归锁搞不定？

看下图

第一个线程过来，拿走了一串钥匙

每次 acquire,就进入一个房间

最后在房间的最里面，没出来。

第二个线程，怎么进入房间呢？

需要第一个线程执行 release 操作，从最里面的房间走出来。

它每走出一个房间，需要 release 一次。将钥匙放到钥匙串上面，归还到门口。

由第二个线程拿走钥匙串才行。

它跟递归函数类似

怎么解决上面的卡住的问题呢？

必须是 4 次 relase

from threading import Thread,RLock
def func(rlock,flag):
    rlock.acquire()  # 第一道锁
    print(flag*10)
    rlock.acquire()  # 第二道锁
    print(flag * 10)
    rlock.acquire()  # 第三道锁
    print(flag * 10)
    rlock.acquire()  # 第四道锁
    print(flag * 10)
    rlock.release()  # 解锁
    rlock.release()
    rlock.release()
    rlock.release()
rlock = RLock()  # 创建递归锁
Thread(target=func,args=(rlock,'*')).start()
Thread(target=func,args=(rlock,'-')).start()

执行输出：

使用递归锁解决科学家吃面的问题

import time
from threading import Thread,RLock
def eat1(noodle_lock,fork_lock,name):  # 吃面
    noodle_lock.acquire()  # 拿到面条的整串钥匙
    print('%s 抢到了面条'%name)
    fork_lock.acquire()  # 拿到叉子的整串钥匙
    print('%s 抢到了叉子'%name)
    print('%s 正在吃面'%name)
    fork_lock.release()  # 叉子解锁
    print('%s归还了叉子'%name)
    noodle_lock.release()  # 面条解锁
    print('%s归还了面条' % name)
def eat2(noodle_lock,fork_lock,name):  #也是吃面，不同的是：先抢叉子，再抢面
    fork_lock.acquire()
    print('%s抢到了叉子' % name)
    time.sleep(0.5)
    noodle_lock.acquire()
    print('%s抢到了面'%name)
    print('%s正在吃面'%name)
    noodle_lock.release()
    print('%s归还了面' % name)
    fork_lock.release()
    print('%s归还了叉子' % name)
noodle_lock = fork_lock = RLock()  # 面条锁和叉子锁，表示一串钥匙
t1 = Thread(target=eat1,args=(noodle_lock,fork_lock,'nazha')).start()
t2 = Thread(target=eat2,args=(noodle_lock,fork_lock,'egon')).start()
t3 = Thread(target=eat1,args=(noodle_lock,fork_lock,'yuan')).start()
t4 = Thread(target=eat2,args=(noodle_lock,fork_lock,'alex')).start()

执行输出：

nazha 抢到了面条

nazha 抢到了叉子

nazha 正在吃面

nazha 归还了叉子

nazha 归还了面条

egon 抢到了叉子

egon 抢到了面

egon 正在吃面

egon 归还了面

egon 归还了叉子

yuan 抢到了面条

yuan 抢到了叉子

yuan 正在吃面

yuan 归还了叉子

yuan 归还了面条

alex 抢到了叉子

alex 抢到了面

alex 正在吃面

alex 归还了面

alex 归还了叉子

4 个人可以欢快的吃面了

什么情况下，需要用到递归锁呢？

有超过一个资源需要锁的时候，使用递归锁

有 2 个屌丝，一个拿着外层钥匙，一个拿着里层钥匙，谁也不想给对方

就出现了死锁

四、信号量

同进程的一样

Semaphore 管理一个内置的计数器，

每当调用 acquire()时内置计数器-1；

调用 release() 时内置计数器+1；

计数器不能小于 0；当计数器为 0 时，acquire()将阻塞线程直到其他线程调用 release()。

实例：(同时只有 5 个线程可以获得 semaphore,即可以限制最大连接数为 5)：

import time
from threading import Thread,Semaphore
def func(sem,i):
    sem.acquire()
    print(i)
    time.sleep(1)
    sem.release()
sem = Semaphore(5)
for i in range(6):
    Thread(target=func,args=(sem,i)).start()

执行输出：

0

1

2

3

4

5

池与信号量

与进程池是完全不同的概念，进程池Pool(4)，最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一堆线程/进程

五、事件

同进程的一样

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用 threading 库中的 Event 对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event 对象中的信号标志被设置为假。如果有线程等待一个 Event 对象, 而这个 Event 对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个 Event 对象的信号标志设置为真,它将唤醒所有等待这个 Event 对象的线程。如果一个线程等待一个已经被设置为真的 Event 对象,那么它将忽略这个事件, 继续执行

event.isSet()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；
event.clear()：恢复event的状态值为False。

例如，有多个工作线程尝试链接 MySQL，我们想要在链接前确保 MySQL 服务正常才让那些工作线程去连接 MySQL 服务器，如果连接不成功，都会去尝试重新连接。那么我们就可以采用 threading.Event 机制来协调各个工作线程的连接操作

连接数据库

import time
import random
from threading import Event,Thread
# 连接数据库
def connect_db(e):
    count = 1  # 初始计数器
    while count < 4:
        print('尝试第%s次检测连接'%count)
        e.wait(0.5)  # 等待0.5,再去执行下面的代码
        # 如果不传参数会一直等到事件为True为止
        # 如果传参数 传一个时间参数,到时间后,
        count += 1  # 加1
        if e.is_set():  # 判断状态是否为True
            print('连接成功')
            break
    else:
        print('连接失败')
def check_conn(e):
    '''检测数据库是否可以连接'''
    time.sleep(random.randint(1,2))  # 等待1~2秒
    e.set()  # 设置状态为True
e = Event()
Thread(target=check_conn,args=(e,)).start()
Thread(target=connect_db,args=(e,)).start()

执行输出：

尝试第 1 次检测连接

尝试第 2 次检测连接

连接成功

结果是不固定的。有 3 种情况

1. check_conn 等待 1 秒，尝试 2 次，连接成功

2. check_conn 等待 1 秒，尝试 3 次，连接成功

3. check_conn 等待 2 秒，尝试 3 次，连接失败

第 3 种情况是必然的，为什么呢？

如果等待 2 秒，e.wait(0.5)执行 4 次，才能达到 2 秒。但是 whlie 循环，不可能达到 4。所以 3 次全部失败。

第 1 种情况，等待 1 秒，e.wait(0.5)必然执行 2 次，才能达到 1 秒，状态变成 True，输出连接成功。

第 2 种情况，是偶然事件。因为 CPU 同一时刻只能执行一个线程。所以失败了 3 次，才输出连接成功

什么时候，用到事件？

你要做一件事情 是有前提的

你就先去处理前提的问题 —— 前提处理好了 把状态设置成True

来控制即将要做的事情可以开始

条件

使得线程等待，只有满足某条件时，才释放 n 个线程

Python提供的Condition对象提供了对复杂线程同步问题的支持。Condition被称为条件变量，除了提供与Lock类似的acquire和release方法外，还提供了wait和notify方法。线程首先acquire一个条件变量，然后判断一些条件。如果条件不满足则wait；如果条件满足，进行一些处理改变条件后，通过notify方法通知其他线程，其他处于wait状态的线程接到通知后会重新判断条件。不断的重复这一过程，从而解决复杂的同步问题。

起 10 个线程

from threading import Condition,Thread
# acquire release
# notify -- wait的行为
# 10线程 wait
# notify(1)
def func(i,con):
    con.acquire()  # 进入锁定池
    con.wait()  # 等待通知,前后必须要有acquire和release
    print(i*'*')
    con.release()  # 解锁
con = Condition()  #条件变量
for i in range(10):
    Thread(target=func,args=(i,con)).start()
while True:
    n = int(input('>>>'))
    con.acquire()  # 加锁
    con.notify(n)  # 通知其他线程,其他处于wait状态的线程接到通知后会重新判断条件,解除wait状态。前后必须要有acquire和release
    con.release()  # 解锁

执行输出：

首先输入 2，显示 1 个*和 0 个星号，也就是空。

再输入 5，分别显示 2~6 个*号

为什么输入 5 的时候，出现 6 个*号？

因为线程池有 10 个，不等长度的*号

输入 2，表示解除 2 个线程的 wait 状态，分别表示 0,1。

那么 5，表示 5 个。分别表示 2,3,4,5,6

从结果上来看，发生数据混乱

con.notify(n) 表示按照要求放开线程

con.notify_all() 表示一次性放开线程

总结：

semaphore 允许同一时刻 n 个线程执行这段代码

event 有一个内部的事件来控制 wait 的行为,控制的是所有的线程

condition 有一个内部的条件来控制 wait 的行为，它可以逐个或者分批次的控制线程的走向

六、定时器

定时器，指定 n 秒后执行某个操作

import time
from threading import Timer
def func():
    print('*'*10)
    print('子线程',time.strftime("%Y-%m-%d %H:%M:%S"))
t =Timer(5,func)  # 要开始一个线程,等到5秒之后才开启并执行
t.start()
print('主进程',time.strftime("%Y-%m-%d %H:%M:%S"))

执行输出：

主进程 2018-05-17 19:41:08

子线程 2018-05-17 19:41:13

虽然可以用 time.sleep(5)完成 5 秒的过程，但是它会阻塞主线程

如果用 timer，就不会，它是异步的。

七、线程队列

queue 队列 ：使用 import queue，用法与进程 Queue 一样

queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.

class queue.Queue(maxsize=0) #先进先出

import queue
q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')
print(q.get())
print(q.get())
print(q.get())
'''
结果(先进先出):
first
second
third
'''

class **queue.LifoQueue(maxsize=0) #last in fisrt out**

import queue
q=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')
print(q.get())
print(q.get())
print(q.get())
'''
结果(后进先出):
third
second
first
'''

class **queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列**

优先级队列

import queue
q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
q.put((20,'a'))
q.put((10,'b'))
q.put((30,'c'))
print(q.get())
print(q.get())
print(q.get())
'''
结果(数字越小优先级越高,优先级高的优先出队):
(10, 'b')
(20, 'a')
(30, 'c')
'''

更多方法说明

Constructor for a priority queue. maxsize is an integer that sets the upperbound limit on the number of items that can be placed in the queue. Insertion will block once this size has been reached, until queue items are consumed. If maxsize is less than or equal to zero, the queue size is infinite.
The lowest valued entries are retrieved first (the lowest valued entry is the one returned by sorted(list(entries))[0]). A typical pattern for entries is a tuple in the form: (priority_number, data).
exception queue.Empty
Exception raised when non-blocking get() (or get_nowait()) is called on a Queue object which is empty.
exception queue.Full
Exception raised when non-blocking put() (or put_nowait()) is called on a Queue object which is full.
Queue.qsize()
Queue.empty() #return True if empty  
Queue.full() # return True if full 
Queue.put(item, block=True, timeout=None)
Put item into the queue. If optional args block is true and timeout is None (the default), block if necessary until a free slot is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Full exception if no free slot was available within that time. Otherwise (block is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (timeout is ignored in that case).
Queue.put_nowait(item)
Equivalent to put(item, False).
Queue.get(block=True, timeout=None)
Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Empty exception if no item was available within that time. Otherwise (block is false), return an item if one is immediately available, else raise the Empty exception (timeout is ignored in that case).
Queue.get_nowait()
Equivalent to get(False).
Two methods are offered to support tracking whether enqueued tasks have been fully processed by daemon consumer threads.
Queue.task_done()
Indicate that a formerly enqueued task is complete. Used by queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.
If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue).
Raises a ValueError if called more times than there were items placed in the queue.
Queue.join() block直到queue被消费完毕

后进先出

import queue
q = queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')
print(q.get())

执行输出：third

先进先出

import queue
q = queue.Queue()
q.put('first')
q.put('second')
q.put('third')
print(q.get())

执行输出：third

优先级队列

import queue
q = queue.PriorityQueue()
q.put((1,'a'))
q.put((2,'z'))
q.put((3,'b'))
print(q.get())

执行输出：(1, ‘a’)

数字越小，优先级越高

有一个场景，你冲了 VIP，你就先买到票

如果有 2 个人，优先级一样呢？

import queue
q = queue.PriorityQueue()
q.put((2,'z'))
q.put((2,'a'))
print(q.get())

执行输出：

(2, ‘a’)

结果是根据 ascii 码的顺序来排序的

八、Python 标准模块—concurrent.futures

https://docs.python.org/dev/library/concurrent.futures.html

#1 介绍
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor：线程池，提供异步调用
ProcessPoolExecutor: 进程池，提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.
#2 基本方法
#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)
回调函数

ProcessPoolExecutor

#介绍
The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls asynchronously. ProcessPoolExecutor uses the multiprocessing module, which allows it to side-step the Global Interpreter Lock but also means that only picklable objects can be executed and returned.
class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None)
An Executor subclass that executes calls asynchronously using a pool of at most max_workers processes. If max_workers is None or not given, it will default to the number of processors on the machine. If max_workers is lower or equal to 0, then a ValueError will be raised.
#用法
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2
if __name__ == '__main__':
    executor=ProcessPoolExecutor(max_workers=3)
    futures=[]
    for i in range(11):
        future=executor.submit(task,i)
        futures.append(future)
    executor.shutdown(True)
    print('+++>')
    for future in futures:
        print(future.result())

ThreadPoolExecutor

#介绍
ThreadPoolExecutor is an Executor subclass that uses a pool of threads to execute calls asynchronously.
class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='')
An Executor subclass that uses a pool of at most max_workers threads to execute calls asynchronously.
Changed in version 3.5: If max_workers is None or not given, it will default to the number of processors on the machine, multiplied by 5, assuming that ThreadPoolExecutor is often used to overlap I/O instead of CPU work and the number of workers should be higher than the number of workers for ProcessPoolExecutor.
New in version 3.6: The thread_name_prefix argument was added to allow users to control the threading.Thread names for worker threads created by the pool for easier debugging.
#用法
与ProcessPoolExecutor相同

map的用法

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2
if __name__ == '__main__':
    executor=ThreadPoolExecutor(max_workers=3)
    # for i in range(11):
    #     future=executor.submit(task,i)
    executor.map(task,range(1,12)) #map取代了for+submit

回调函数

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from multiprocessing import Pool
import requests
import json
import os
def get_page(url):
    print('<进程%s> get %s' %(os.getpid(),url))
    respone=requests.get(url)
    if respone.status_code == 200:
        return {'url':url,'text':respone.text}
def parse_page(res):
    res=res.result()
    print('<进程%s> parse %s' %(os.getpid(),res['url']))
    parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))
    with open('db.txt','a') as f:
        f.write(parse_res)
if __name__ == '__main__':
    urls=[
        'https://www.baidu.com',
        'https://www.python.org',
        'https://www.openstack.org',
        'https://help.github.com/',
        'http://www.sina.com.cn/'
    ]
    # p=Pool(3)
    # for url in urls:
    #     p.apply_async(get_page,args=(url,),callback=pasrse_page)
    # p.close()
    # p.join()
    p=ProcessPoolExecutor(3)
    for url in urls:
        p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果

import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print(i*'*')
    time.sleep(1)
thread_pool = ThreadPoolExecutor(5)  #创建一个最大可容纳2个任务的线程池
thread_pool.submit(func,1)  # 异步提交任务,往线程池里面加入一个任务

执行输出：

*

添加 10 个任务

import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print(i*'*')
    time.sleep(1)
thread_pool = ThreadPoolExecutor(5)  #创建一个最大可容纳2个任务的线程池
for i in range(6):
    thread_pool.submit(func,i)  # 异步提交任务,往线程池里面加入一个任务
print('wahaha')

执行输出：

*

**

wahaha

thread_pool.submit 相当于之前线程的 apply_async，表示异步

由于等到了 1 秒，所以优先输出 wahaha

如果想最后打印 wahaha 呢?

import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print(i*'*')
    time.sleep(1)
thread_pool = ThreadPoolExecutor(5)  #创建一个最大可容纳2个任务的线程池
for i in range(6):
    thread_pool.submit(func,i)  # 异步提交任务,往线程池里面加入一个任务
thread_pool.shutdown()  # 相当于进程池的pool.close()+pool.join()操作
print('wahaha')

执行输出：

*

**

wahaha

import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print(i*'*')
    time.sleep(1)
    return i**2
thread_pool = ThreadPoolExecutor(5)  #创建一个最大可容纳2个任务的线程池
ret_lst = []
for i in range(6):
    ret = thread_pool.submit(func,i)  # 异步提交任务,往线程池里面加入一个任务
    ret_lst.append(ret)
thread_pool.shutdown()  # 相当于进程池的pool.close()+pool.join()操作
for ret in ret_lst:
    print(ret.result())  # 取得结果
print('wahaha')

执行输出：

*

**

0

1

4

9

16

25

wahaha

到目前为止和 ThreadPool 的区别

回调函数

import time
from concurrent.futures import ThreadPoolExecutor
def func(i):
    print(i*'*')
    time.sleep(1)
    return i**2
def callback(arg):  #回调函数
    print(arg.result()*'-')  # 取得结果,并乘以-
thread_pool = ThreadPoolExecutor(5)  #创建一个最大可容纳2个任务的线程池
for i in range(6):
    ret = thread_pool.submit(func,i).add_done_callback(callback)  # 异步提交任务,执行回调函数
thread_pool.shutdown()  # 相当于进程池的pool.close()+pool.join()操作
print('wahaha')

执行输出：

*

**

-

wahaha

未来建议使用 concurrent 里面的 pooll

做效率测试，比较方便

什么情况下，使用进程池/线程池 ？

当内存不需要共享，且高计算的时候 用进程

当内存需要共享，且高 IO 的时候 用线程

当并发很大的时候

多进程 ： 多个任务 —— 进程池 ：cpu个数、cpu个数+1

多线程 ：多个任务 —— 线程池 ：cpu个数*5

4 核 CPU，可以开 4~5 个进程，开 20 条线程/进程。最终可以开 80~100 个任务

最美的程序，不是单一使用。而是复合使用

比如开多个进程，每个进程，开多个线程。

当后面学到协程，并发（qps）能达到 5 万

锁的概念，面试，会经常问到，还有队列

真正工作过程中，锁会用到，其他的，很少用。

还有进程池/线程池也会用到

明日默写：

import time
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
def func(i):
    print(i*'*')
    time.sleep(1)
    return i**2
def callb(arg):
    print(arg.result()*'-')
if __name__ == '__main__':
    thread_pool = ThreadPoolExecutor(5)
    for i in range(10):
        thread_pool.submit(func,i).add_done_callback(callb)   # 相当于apply_async
    thread_pool.shutdown()           # close+join
    print('wahaha')