python多进程、多线程

1、python多线程

使用一个例子来学习多线程。建议自己敲一遍。
python多线程是通过threading模块的Thread实现。
创建线程对象 t = thread.Thread()
启动线程 t.start()

import threading
import timedef say(name):print("你好%s at %s"%(name, time.ctime()))time.sleep(2)print("结束%s at %s"%(name, time.ctime()))def listen(name):print("你好%s at %s"%(name, time.ctime()))time.sleep(4)print("结束%s at %s"%(name, time.ctime()))if __name__=='__main__':t1 = threading.Thread(target=say, args = ('tony',))t1.start()t2 = threading.Thread(target = listen, args = ('simon',))t2.start()print("程序结束=============")

你好tony at Wed Mar 24 17:58:32 2021
你好simon at Wed Mar 24 17:58:32 2021
程序结束=============
结束tony at Wed Mar 24 17:58:34 2021
结束simon at Wed Mar 24 17:58:36 2021

可以看出主线程不是在两个子线程运行结束后退出的。这是因为主线程和两个子线程是同时跑的，但是子线程跑完后，主线程才会退出。

但是有时候我们想要子线程跑完后，再继续跑主线程，这个时候就引入了join

import threading
import timedef say(name):print("你好%s at %s"%(name, time.ctime()))time.sleep(2)print("结束%s at %s"%(name, time.ctime()))def listen(name):print("你好%s at %s"%(name, time.ctime()))time.sleep(4)print("结束%s at %s"%(name, time.ctime()))if __name__=='__main__':t1 = threading.Thread(target=say, args = ('tony',))t1.start()t2 = threading.Thread(target = listen, args = ('simon',))t2.start()t1.join()t2.join()print("程序结束=============")

你好tony at Wed Mar 24 18:01:51 2021
你好simon at Wed Mar 24 18:01:51 2021
结束tony at Wed Mar 24 18:01:53 2021
结束simon at Wed Mar 24 18:01:55 2021
程序结束=============

主线程最后执行print操作，并退出。但是如果不加join，主线程执行打印，但是主线程还没有结束，还需要等待子线程结束后，主线程才结束。

上面的子线程是非守护子线程，默认的子线程都是主线程的非守护子线程。有时候我们需要当主线程结束，不管主线程有没有结束，子线程都要跟随主线程一起退出，这个时候就需要守护线程。

如果某个线程是守护线程，那么主线程就不需要等待这个子线程，当其他非守护线程运行结束后，主线程就退出。
下面我们看个例子，设置t2为守护进程。

import threading
import timedef say(name):print("你好%s at %s"%(name, time.ctime()))time.sleep(2)print("结束%s at %s"%(name, time.ctime()))def listen(name):print("你好%s at %s"%(name, time.ctime()))time.sleep(4)print("结束%s at %s"%(name, time.ctime()))if __name__=='__main__':t1 = threading.Thread(target=say, args = ('tony',))t1.start()t2 = threading.Thread(target = listen, args = ('simon',))t2.setDaemon(True)t2.start()print("程序结束=============")

你好tony at Wed Mar 24 18:07:47 2021
你好simon at Wed Mar 24 18:07:47 2021
程序结束=============
结束tony at Wed Mar 24 18:07:49 2021

t2进程是sleep 4s，主线程不等待t2进程结束，就退出了。
thread的一些方法：

join()：在子线程完成运行之前，这个子线程的父线程将一直被阻塞。

setDaemon(True) 在start()之前设置，这个方法和join是相仿的。

run():线程被cpu调度后自动执行线程对象的run方法

start() 启动线程活动

isAlibe() 返回线程是否活动的。

上面的进程，如果分开跑，需要6s，但是实际上用了4s，说明性能提升了，但是这种提升是cpu并发实现的提升，也就是cpu线程切换（多道技术）带来的，而不是cpu的并行执行。

并发：指一个系统具有处理多个任务的能力（cpu切换，多道技术）
并行：指一个系统有同时处理多个任务的能力（cpu同时处理）
并行是并发的一种情况，子集。

python 多线程不能实现真正的并行操作，是因为GIL（全局解释器锁）

我们对任务进行分类：

IO密集型（各个线程都会各种的等待，如果有等待，切换线程是比较核是的），也可以采用多线程+协程

计算密集型（线程在计算过程中没有等待，这时候没必要做切换）
多线程学习传送门
一篇文章搞懂Python多线程简单实现和GIL

2.多线程同步锁、死锁和递归锁

如果python多个线程要用到相同的数据，就会存在资源争用和锁的问题。

2.1 同步锁

通常用来实现对共享资源的同步访问。为每一个共享资源创建一个Lock对象。如果需要方位该资源，调用acquire方法来湖区对象（如果其他线程已经获取了该锁，则当前线程需要等待其被释放），使用release方法释放锁。
先看下面的案例

num = 100def fun_sub():global numnum2 = numtime.sleep(0.001)num = num2 - 1if __name__ == '__main__':print('开始测试同步锁 at %s' % time.ctime())thread_list = []for thread in range(100):t = threading.Thread(target = fun_sub)t.start()thread_list.append(t)for t in thread_list:t.join()print('num is %d'%num)print('结束测试同步锁 at %s'%time.ctime())

num is 72
结束测试同步锁 at Wed Mar 24 22:06:38 2021

上面的例子，我们每个线程从公共资源num变量执行减1操作，正常情况下，等到代码执行结束，得到的num为0。但是结果却是72，分析下代码执行流程。

因为GIL，只有一个线程（假设线程1）拿到了num资源，然后把变量赋值给num2，sleep 0.001秒，这时候num=100。
第一个线程sleep的时候，这个线程会做yield操作，就是cpu切换给别的线程（假设线程2拿到GIL，获得cpu使用权），线程2拿到和线程1一样的num，返回赋值给这时候num有可能还是100，然后sleep，这个时候num还是100.
线程2 sleep的时候，又要yield操作，假设线程3拿到num，执行上面的操作，num还有可能是100.
等到后面cpu重新切换给线程1，线程2，线程3执行时，他们执行减1操作后，其实得到的num都是99，不是顺序递减的。
其他线程操作如上

所以实际的运行过程并不是我们想象的按顺序减，这个时候就需要python的同步锁了，同一时间只能放一个线程来操作num变量，减1后，后面的线程操作来操作num。
使用同步锁，一次只有一个线程操作同享资源。

num = 100def fun_sub():global numlock.acquire()print('---------加锁------', t.name)num2 = numtime.sleep(0.001)num = num2 - 1lock.release()print('--------释放锁----------', num)if __name__ == '__main__':print('开始测试同步锁 at %s' % time.ctime())lock = threading.Lock() #创建锁thread_list = []for thread in range(100):t = threading.Thread(target = fun_sub)t.start()thread_list.append(t)print('join')for t in thread_list:t.join()print('num is %d'%num)print('结束测试同步锁 at %s'%time.ctime())

num is 0
结束测试同步锁 at Wed Mar 24 18:35:09 2021

2.2 python死锁

介绍下死锁怎么产生的：

A拿了一个苹果
B拿了一个香蕉
A现在想再拿个香蕉，就在等待B释放这个香蕉
B同时想要再拿个苹果，这时候就等待A释放苹果
这样就是陷入了僵局，这就是生活中的死锁

python中在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁，因为系统判断这部分资源都正在使用，所有这两个线程在无外力作用下将一直等待下去。下面是一个死锁的例子：

lock_apple = threading.Lock()
lock_banana = threading.Lock()class MyThread(threading.Thread):def __init__(self):threading.Thread.__init__(self)def run(self):self.fun1()self.fun2()def fun1(self):lock_apple.acquire()print("线程 %s, 想拿： %s--%s"%(self.name,"苹果",time.ctime()))lock_banana.acquire()print("线程%s, 想拿： %s--%s"%(self.name,"香蕉",time.ctime()))lock_banana.release()lock_apple.release()def fun2(self):lock_banana.acquire()print("线程%s, 想拿： %s--%s"%(self.name,"香蕉",time.ctime()))time.sleep(0.1)lock_apple.acquire()print("线程 %s, 想拿： %s--%s"%(self.name,"苹果",time.ctime()))lock_apple.release()lock_banana.release()if __name__ == "__main__":for i in range(0, 10):my_thread = MyThread()my_thread.start()

线程 Thread-1, 想拿： 苹果--Wed Mar 24 22:42:14 2021
线程Thread-1, 想拿： 香蕉--Wed Mar 24 22:42:14 2021
线程Thread-1, 想拿： 香蕉--Wed Mar 24 22:42:14 2021
线程 Thread-2, 想拿： 苹果--Wed Mar 24 22:42:14 2021

代码处理流程：

fun1中，线程1先拿了苹果，然后拿了香蕉，然后释放香蕉和苹果，然后再在fun2中又拿了香蕉，sleep 0.1秒。
在线程1的执行过程中，线程2进入了，因为苹果被线程1释放了，线程2这时候获得了苹果，然后想拿香蕉
这时候就出现问题了，线程一拿完香蕉之后想拿苹果，返现苹果被线程2拿到了，线程2拿到苹果执行，想拿香蕉，发现香蕉被线程1持有了
双向等待，出现死锁，代码执行不下去了

2.3 Python递归锁RLock

import threading
import timelock = threading.RLock()  #递归锁class MyThread(threading.Thread):def __init__(self):threading.Thread.__init__(self)def run(self):self.fun1()self.fun2()def fun1(self):lock.acquire()  # 如果锁被占用,则阻塞在这里,等待锁的释放print ("线程 %s , 想拿: %s--%s" %(self.name, "苹果",time.ctime()))lock.acquire()print ("线程 %s , 想拿: %s--%s" %(self.name, "香蕉",time.ctime()))lock.release()lock.release()def fun2(self):lock.acquire()print ("线程 %s , 想拿: %s--%s" %(self.name, "香蕉",time.ctime()))time.sleep(0.1)lock.acquire()print ("线程 %s , 想拿: %s--%s" %(self.name, "苹果",time.ctime()))lock.release()lock.release()if __name__ == "__main__":for i in range(0, 10):  #建立10个线程my_thread = MyThread()  #类继承法是python多线程的另外一种实现方式my_thread.start()

上面我们用一把递归锁，就解决了多个同步锁导致的死锁问题。大家可以把RLock理解为大锁中还有小锁，只有等到内部所有的小锁，都没有了，其他的线程才能进入这个公共资源。

3.多进程

3.1 多进程模块`multiprocessing`

from multiprocessing import Process# sample
def fun1(name):print("测试%s多进程"%name)if __name__ == '__main__':process_list = []for i in range(5):p = Process(target = fun1, args=('python',))p.start()process_list.append(p)for i in process_list:p.join()print('结束测试')

使用类继承方法实现多进程

## inherit
class MyProcess(Process):def __init__(self, name):super(MyProcess, self).__init__()self.name = namedef run(self):print('测试%s多进程'%self.name)if __name__ =='__main__':process_list = []for i in range(5):p = MyProcess('Python')p.start()process_list.append(p)for i in process_list:p.join()print('测试结束')

3.2 python多线程通信

进程是系统独立调度核分配系统资源（CPU、内存）的基本单位，进程之间是相互独立的，每启动一个新的进程相当于把数据进行了一次克隆，子进程里的数据修改无法影响到主进程中的数据，不同子进程之间的数据也不能共享，这是多进程在使用中与多线程最明显的区别。但是难道Python多进程中间难道就是孤立的吗？当然不是，python也提供了多种方法实现了多进程中间的通信和数据共享（可以修改一份数据）。

3.2.1进程队列 Queue

通过Queue获取子进程中put的数据，实现进程间的通信。

from multiprocessing import Process, Queuedef fun1(q, i):print('子进程%s 开始put数据'%i)q.put('我是%s通过Queue通信'%i)if __name__ == '__main__':q = Queue()process_list = []for i in range(3):p = Process(target=fun1, args = (q, i))p.start()process_list.append(p)for i in process_list:p.join()print('主进程获取Queue数据')print(q.get())print(q.get())print(q.get())print('结束测试')

3.2.2 管道Pipe

管道Pipe和Queue的作用大致差不多，也是实现进程间的通信

from multiprocessing import Process, Pipe
def fun1(conn):print('子进程发送消息：')conn.send('你好主进程')print('子进程接收消息：')print(conn.recv())conn.close()if __name__ == '__main__':conn1, conn2 = Pipe()p = Process(target = fun1, args = (conn2, ))p.start()print('主进程接受消息:')print(conn1.recv())print('主进程发送消息： ')conn1.send("你好子进程")p.join()print('测试结束')

主进程接受消息:
子进程发送消息：
子进程接收消息：
你好主进程
主进程发送消息：
你好子进程
测试结束

3.2.3 Manager

Queue和Pipe只是实现了数据交互，并没实现数据共享，即一个进程去更改另一个进程的数据。那么就要用到Managers。

from multiprocessing import Process, Manager
def fun1(dic, lis, index):dic[index] = 'a'dic['2'] = 'b'lis.append(index)if __name__ == '__main__':with Manager() as manager:dic = manager.dict()l = manager.list(range(5))process_list = []for i in range(10):p = Process(target = fun1, args = (dic, l, i))p.start()process_list.append(p)for res in process_list:res.join()print(dic)print(l)

{0: 'a', '2': 'b', 1: 'a', 2: 'a', 3: 'a', 4: 'a', 5: 'a', 6: 'a', 9: 'a', 8: 'a', 7: 'a'}
[0, 1, 2, 3, 4, 0, 1, 2, 3, 4, 5, 6, 9, 8, 7]

可以看到主进程定义了一个字典和一个列表，在子进程中，可以添加和修改字典的内容，在列表中插入新的数据，实现进程间的数据共享，即可以共同修改同一份数据.

3.2.4 进程池

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。就是固定有几个进程可以使用。

3.2.4 进程池map方法

import os
import PIL from multiprocessing import Pool
from PIL import ImageSIZE = (75,75)
SAVE_DIRECTORY = \'thumbs\'def get_image_paths(folder):return (os.path.join(folder, f) for f in os.listdir(folder) if \'jpeg\' in f)def create_thumbnail(filename): im = Image.open(filename)im.thumbnail(SIZE, Image.ANTIALIAS)base, fname = os.path.split(filename) save_path = os.path.join(base, SAVE_DIRECTORY, fname)im.save(save_path)if __name__ == \'__main__\':folder = os.path.abspath(\'11_18_2013_R000_IQM_Big_Sur_Mon__e10d1958e7b766c3e840\')os.mkdir(os.path.join(folder, SAVE_DIRECTORY))images = get_image_paths(folder)pool = Pool()pool.map(creat_thumbnail, images) #关键点，images是一个可迭代对象pool.close()pool.join()

上边这段代码的主要工作就是将遍历传入的文件夹中的图片文件，一一生成缩略图，并将这些缩略图保存到特定文件夹中。这我的机器上，用这一程序处理 6000 张图片需要花费 27.9 秒。 map 函数并不支持手动线程管理，反而使得相关的 debug 工作也变得异常简单。

map在爬虫的领域里也可以使用，比如多个URL的内容爬取，可以把URL放入元祖里，然后传给执行函数。