任何一门语言都需要有多任务处理能力，Python自然也一样，有很多人诟病Python慢，Python慢也只是多线程的时候慢，如果你还没有使用多线程就说Python慢，那就是你的问题了，话说回来，为什么Python的多线程会慢呢，是因为CPython解释器的GIL锁的问题，Python的多线程直接调用的操作系统的多线程，也就是说Python的多线程是真的多线程，不是虚拟出来的，CPython在解释Python代码的过程中理论上单线程的，CPython会在多个操作系统线程上解释Python代码，这是Python慢的根本原因。如果你使用其他Python解释器就不会有这种问题，CPython也尝试去掉GIL锁，但是效率反而没有现在高。
说的有点远，对于Python来说，实现多任务的方式有三种，多进程，多线程和协程，只有多进程能够真正的并行，多线程和协程都是并发。

1 进程

python的内置multiprocessing包提供了调用多进程的接口。

1.1 基本应用

import time
import multiprocessingdef task(name):while True:print(f"Hello {name}")time.sleep(1)def main():p1 = multiprocessing.Process(target=task, args=("xiaoming",), name="xiaoming")p2 = multiprocessing.Process(target=task, args=("xiaohong",), name="xiaohong")p1.start()if p1.is_alive():print(f"{p1.name}-{p1.pid}")p2.start()if p2.is_alive():print(f"{p2.name}-{p2.pid}")if __name__ == '__main__':print(f"主进程开始：{multiprocessing.process.current_process().pid}-{multiprocessing.process.current_process().pid}")main()

这是最基本的应用，同一个任务需要两个进程共同执行。

1.2 子进程做为主进程的守护进程

有一些场景可能需要子进程守护主进程，比如，一款带北京音乐的游戏，游戏启动后背景音乐要跟着启动，游戏停止了，背景音乐自然也要停止，不能游戏停止了背景音乐还在继续。

class Game:def __init__(self, name):self.name = namedef screen(self):while True:print(f"玩{self.name}中。。。。")time.sleep(1)def bg_music(self):while True:print("背景音乐播放中。。。。")time.sleep(2)def main():game = Game("CS1.5", True)p_game = multiprocessing.Process(target=game.screen, daemon=True)p_mucis = multiprocessing.Process(target=game.bg_music, daemon=True)p_game.start()p_mucis.start()time.sleep(10)print("老婆回家了，赶紧收电脑")if __name__ == '__main__':main()

上面的例子有三条进程，主进程模拟玩了10秒钟游戏后停止，游戏子进程和背景音乐子进程是主进程的守护进程，主进程停止后，自动停止。

1.3 操作进程的常用方法

属性

p.name 
p.pid 

方法

p.is_alive()  # 返回True False 
p.start()     # 启动进程
p.join()    # 主进程等待子进程执行结束
p.terminate()    # 终止进程
p.kill()      # 杀死进程

1.4 子进程拥有独立的内存空间

进程是操作系统分配资源的最小单位，每个进程拥有独立的地址空间。

import multiprocessingname = "哈利波特"def show():print(name)print(f"子进程{multiprocessing.process.current_process().pid}内，name的内存地址为{id(name)}")def main():p1 = multiprocessing.Process(target=show)p2 = multiprocessing.Process(target=show)p1.start()p2.start()print(f"主进程内：name的内存地址：{id(name)}")if __name__ == '__main__':main()

为了区分明显，这里先剧透一下多线程，同一个进程中的多个线程共享同一个内存地址空间。

import threadingname = "哈利波特"def show():print(name)print(f"子线程{threading.current_thread().ident}内，name的内存地址为{id(name)}")def main():t1 = threading.Thread(target=show)t2 = threading.Thread(target=show)t1.start()t2.start()print(f"主线程内：name的内存地址：{id(name)}")if __name__ == '__main__':main()

1.5 进程间的通信

前面的例子基本上都是一个进程就做一件事情，但是在实际的开发过程中，大多数都是进程之间配合完成工作的，常用的生产者消费者模型，多进程之间的通信可以使用队列实现。

import time
import multiprocessing
import randomdef product(q, delay):while True:if q.full():print("生产力饱和，减慢生产速度")delay = 10else:x = random.randint(1, 100)print(f"生产者生产{x}")q.put(x)delay = 1time.sleep(delay)def consume(q, delay):while True:if q.empty():print("消费速度快，减速消费")delay = 10else:result = q.get()print(f"消费者消费: {result}")delay=1time.sleep(delay)def main():q = multiprocessing.Queue(maxsize=10)p_pro = multiprocessing.Process(target=product, args=(q, 1))p_con = multiprocessing.Process(target=consume, args=(q, 1))p_pro.start()p_con.start()if __name__ == '__main__':main()

注意这里的队列不要使用queue,Queue，这个队列不支持多线程，会报错

1.6 进程池

创建进程销毁进程都会占用系统资源，在不清楚需要多少进程的情况下，比如web请求，不知道什么时候来的请求会多，这种情况下可以使用池化技术，创建一个任务队列，所有的任务都在任务队列中排队，进程池每次处理指定数量的任务，进程循环利用，省去了创建和销毁的性能开销。
标准库中的这两个模块提供了进程池和线程池的支持：
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

2 线程

进程是操作系统调度资源的最小单元，线程是CPU调度的最小单元，同一个进程中可以有多个线程，多个线程共享同一片地址空间

2.1 基本使用

threading模块的Thread类提供Python多线程高级别API

import threadingdef sing():while True:print("singing")def dance():while True:print("dancing")def main():threading.Thread(target=sing).start()threading.Thread(target=dance).start()if __name__ == '__main__':main()

线程与进程相同的定义就不重复，可以模仿进程操作

2.2 本地线程

本地线程就是每个线程拥有一个隔离的内存空间。多线程不会互相干扰‘

2.3 线程锁

我只是模拟了一下多线程争夺全局变量的问题，实际中不会用多线程做这种减法

import time
import threadingcount = 10def decr():global counttemp = count # 模拟延时，0.01s对于人类来说非常短，但是对于CPU来讲可以做很多事情，# 当遇到IO阻塞后，操作系统会切换到别的线程，这时候，count还没有减去1，# 所以10个线程拿到的count数据都是10，别减了10次还是9 ，这是明显不对的time.sleep(0.01) count = temp - 1print(f"==当前线程id{threading.current_thread().ident}==count = {count}\n")ts = []for i in range(10):t = threading.Thread(target=decr)t.start()ts.append(t)for t in ts:t.join()  # 主线程阻塞等待子线程执行结束后再结束
"""
==当前线程id2332==count = 9
==当前线程id12000==count = 9
==当前线程id8972==count = 9
==当前线程id13424==count = 9
==当前线程id12016==count = 9
==当前线程id3620==count = 9
==当前线程id12740==count = 9
==当前线程id10176==count = 9
==当前线程id12116==count = 9
==当前线程id1052==count = 9
"""

使用锁，一个线程对一段代码上锁后，只有这个线程解锁后，其他线程才能继续操作

import time
import threadingcount = 10lock = threading.Lock()def decr():try:lock.acquire() # 获取锁global counttemp = counttime.sleep(0.01)count = temp - 1print(f"==当前线程id{threading.current_thread().ident}==count = {count}\n")finally:lock.release() # 释放锁 支持with上下文管理器ts = []for i in range(10):t = threading.Thread(target=decr)t.start()ts.append(t)for t in ts:t.join()  # 主线程阻塞等待子线程执行结束后再结束"""
==当前线程id10828==count = 9
==当前线程id1044==count = 8
==当前线程id4116==count = 7
==当前线程id1624==count = 6
==当前线程id660==count = 5
==当前线程id2748==count = 4
==当前线程id6552==count = 3
==当前线程id6540==count = 2
==当前线程id2836==count = 1
==当前线程id7304==count = 0
"""

2.4 死锁

import time
import threading# 创建两把锁
lockA = threading.Lock()
lockB = threading.Lock()a = 0
b = 0def taskA():global a, blockA.acquire()temp = aprint("变量a上锁")lockB.acquire()temp_ = bprint("变量b上锁")lockB.release()print("变量b释放锁")lockA.release()print("变量a释放锁")# taskA 抢a锁的时候，刚好taskB抢b锁
def taskB():global a, blockB.acquire()time.sleep(1)temp = bprint("变量b上锁")lockA.acquire()temp_ = aprint("变量a上锁")lockB.release()print("变量a释放锁")lockA.release()print("变量b释放锁")def main():taskA()taskB()ts = []for i in range(10):t = threading.Thread(target=main)t.start()ts.append(t)for t in ts:t.join()  # 主线程阻塞等待子线程执行结束后再结束

2.5 递归锁（解决死锁）

当两个线程同时争一把锁的情况下，就会出现死锁，程序会一直阻塞住，不往下运行了，所以程序中一定要避免使用死锁，递归锁可以解决这个问题，RLock本身有一个计数器，如果碰到acquire，那么计数器+1，如果计数器大于0，那么其他线程无法获取锁，如果碰到release，计数器-1

import time
import threading# 创建两把锁
lockR = threading.RLock()a = 0
b = 0def taskA():global a, blockR.acquire()temp = aprint("变量a上锁")lockR.acquire()temp_ = bprint("变量b上锁")lockR.release()print("变量b释放锁")lockR.release()print("变量a释放锁")# taskA 抢a锁的时候，刚好taskB抢b锁
def taskB():global a, blockR.acquire()time.sleep(1)temp = bprint("变量b上锁")lockR.acquire()temp_ = aprint("变量a上锁")lockR.release()print("变量a释放锁")lockR.release()print("变量b释放锁")def main():taskA()taskB()ts = []for i in range(10):t = threading.Thread(target=main)t.start()ts.append(t)for t in ts:t.join()  # 主线程阻塞等待子线程执行结束后再结束

2.6 信号量

信号量本身也是一种锁，在同一时刻，只能有设定数量的线程执行操作

import threading
import timesem = threading.Semaphore(5)def task():with sem:print("Hello")time.sleep(2)ts = []for i in range(100):t = threading.Thread(target=task)t.start()ts.append(t)for t in ts:t.join()

上面的简单示例没2s打印5次hello。

2.7 事件对象

生活中的小常识，必须要切好菜之后才能炒菜，菜没切好就要等着，对于线程来说就是阻塞住。
Event对象提供了简单的线程通信方式，

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

import threading
import timeevent = threading.Event()def qiecai():print("开始切菜")time.sleep(5)event.isSet() or event.set()def zuofan():print("等待切菜。。。")event.wait()print("菜切好了，开始炒菜")time.sleep(3)print("出锅")event.clear()t1 = threading.Thread(target=qiecai)
t2 = threading.Thread(target=zuofan)t1.start()
t2.start()t1.join()
t2.join()

2.8 条件对象

acquire(*args)
请求底层锁。此方法调用底层锁的相应方法，返回值是底层锁相应方法的返回值。release()
释放底层锁。此方法调用底层锁的相应方法。没有返回值。wait(timeout=None)
等待直到被通知或发生超时。如果线程在调用此方法时没有获得锁，将会引发 RuntimeError 异常。这个方法释放底层锁，然后阻塞，直到在另外一个线程中调用同一个条件变量的 notify() 或 notify_all() 唤醒它，或者直到可选的超时发生。一旦被唤醒或者超时，它重新获得锁并返回。当提供了 timeout 参数且不是 None 时，它应该是一个浮点数，代表操作的超时时间，以秒为单位（可以为小数）。当底层锁是个 RLock ，不会使用它的 release() 方法释放锁，因为当它被递归多次获取时，实际上可能无法解锁。相反，使用了 RLock 类的内部接口，即使多次递归获取它也能解锁它。 然后，在重新获取锁时，使用另一个内部接口来恢复递归级别。返回 True ，除非提供的 timeout 过期，这种情况下返回 False。在 3.2 版更改: 很明显，方法总是返回 None。wait_for(predicate, timeout=None)
等待，直到条件计算为真。 predicate 应该是一个可调用对象而且它的返回值可被解释为一个布尔值。可以提供 timeout 参数给出最大等待时间。这个实用方法会重复地调用 wait() 直到满足判断式或者发生超时。返回值是判断式最后一个返回值，而且如果方法发生超时会返回 False 。忽略超时功能，调用此方法大致相当于编写:while not predicate():cv.wait()
因此，规则同样适用于 wait() ：锁必须在被调用时保持获取，并在返回时重新获取。 随着锁定执行判断式。notify(n=1)
默认唤醒一个等待这个条件的线程。如果调用线程在没有获得锁的情况下调用这个方法，会引发 RuntimeError 异常。这个方法唤醒最多 n 个正在等待这个条件变量的线程；如果没有线程在等待，这是一个空操作。当前实现中，如果至少有 n 个线程正在等待，准确唤醒 n 个线程。但是依赖这个行为并不安全。未来，优化的实现有时会唤醒超过 n 个线程。注意：被唤醒的线程并没有真正恢复到它调用的 wait() ，直到它可以重新获得锁。 因为 notify() 不释放锁，其调用者才应该这样做。notify_all()
唤醒所有正在等待这个条件的线程。这个方法行为与 notify() 相似，但并不只唤醒单一线程，而是唤醒所有等待线程。如果调用线程在调用这个方法时没有获得锁，会引发 RuntimeError 异常。notifyAll 方法是此方法的已弃用别名。

import threading
import timecount = 500
con = threading.Condition()class Producer(threading.Thread):# 生产者函数def run(self):global countwhile True:if con.acquire():# 当count 小于等于1000 的时候进行生产if count > 1000:con.wait()else:count = count + 100msg = self.name + ' produce 100, count=' + str(count)print(msg)# 完成生成后唤醒waiting状态的线程，# 从waiting池中挑选一个线程，通知其调用acquire方法尝试取到锁con.notify()con.release()time.sleep(1)class Consumer(threading.Thread):# 消费者函数def run(self):global countwhile True:# 当count 大于等于100的时候进行消费if con.acquire():if count < 100:con.wait()else:count = count - 5msg = self.name + ' consume 5, count=' + str(count)print(msg)con.notify()# 完成生成后唤醒waiting状态的线程，# 从waiting池中挑选一个线程，通知其调用acquire方法尝试取到锁con.release()time.sleep(1)def test():for i in range(2):p = Producer()p.start()for i in range(5):c = Consumer()c.start()if __name__ == '__main__':test()