手把手教你实现Python多进程多线程

2020年12月11日 1条评论 9,270次阅读 9人点赞

简介

在程序设计当中会碰到任务并发和并行处理的情况，此时就需要使用多进程（process）和多线程（thread）来加速程序执行效率。

进程是操作系统能独立调度的最小单位，而线程是进程中可并发执行的单元；
一个应用程序至少包括1个进程，而1个进程包括1个或多个线程；
每个进程在执行过程中拥有独立的内存单元，而一个进程的多个线程在执行过程中共享内存。

一个形象化的理解如下图所示：

《手把手教你实现Python多进程多线程》

计算机的CPU就像一个工厂，而进程就是工厂里面的车间，线程就是车间里面的工人。因此我们能够发现，多核CPU对应多个工厂，这些工厂可以从事不同的生产任务。比如有炼铁的、炼油的、食品加工等等。在一个工厂中，又可以包含一个或多个车间（进程），但是由于电力资源有限，同一时间就只能有一个车间在运行。换句话说，对于单核CPU，同一时间只能处理一个进程，其他进程处在等待状态。而一个车间中，可以有一个或多个工人（线程）协同工作，他们共享内部资源，共同完成任务。

对于多进程而言，就是利用多核CPU的优势实现多个任务的同时进行；对于多线程而言，就是利用多个线程（工人）完成一个进程（车间）中的任务。

多进程、多线程如何选择

既然多进程和多线程都能加速执行效率，那么选择哪个更好呢？

这里直接上结论：

对于计算密集型任务，多进程效率更高；
对I/O（输入输出）密集型任务，多线程效率更高（多进程也能实现，只是效率低、消耗内存较多）。

计算密集型主要是指程序任务需要经常调用CPU的运算器，比如数值计算等，而I/O密集型是指程序任务以文件输入输出为主，比如文件操作、爬虫等。为什么是这样呢？其实也不难理解。对于IO密集型操作，大部分消耗时间其实是等待时间，在等待时间中CPU是不需要工作的，那你在此期间提供双CPU资源也是利用不上的，相反对于CPU密集型代码，2个CPU干活肯定比一个CPU快很多。那么为什么多线程会对IO密集型代码有用呢？这时因为python碰到等待会释放GIL供新的线程使用，实现了线程间的切换。

多进程创建方法Process类

Python内置的 multiprocessing 模块提供了创建多进程的方法。

我们首先来看下 Process 类的构造方法为：

class multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

1	class multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

参数说明如下：

target表示调用对象，一般为函数，也可以是类；
args表示调用对象的位置参数元组；
kwargs表示调用对象的字典；
name为进程的别名；
group参数不使用，可忽略。

提供的方法如下：

is_alive()：返回进程是否在运行；
join([timeout])：阻塞进程，直到进程执行完成或超时或进程被终止；
run()：代表进程执行的任务函数，可被重写。start()调用run()方法，如果实例化进程时未指定传入target，start则执行默认run()方法；
start()：激活进程；
terminate()：终止进程。

属性如下：

authkey：字节码，进程的准密钥；
daemon：父进程终止后自动终止，且不能产生新进程，必须在start()之前设置；
exitcode：退出码，进程在运行时为None，如果为-N，就表示被信号N结束；
name：获取进程名称；
pid：进程id。

首先我们来介绍最基本的 Process 类，其创建进程的方式有以下两种：

直接创建一个 Process 类的实例，并传入目标函数；
自定义一个类并继承 Process 类，重写其 __init__() 方法和 run() 方法。

直接创建Process类实例

一个简单的例子创建多进程：

from multiprocessing import Process

def work(i):
    print('调用第{}个进程'.format(i))

if __name__ == '__main__':
    # 创建10个进程
    for i in range(10):
        p = Process(target=work, args=(i,))  # target传入目标函数，args传入目标函数所需参数
        p.start()  # 启动进程

from multiprocessing import Process

def work(i):

print('调用第{}个进程'.format(i))

if __name__ == '__main__':

# 创建10个进程

for i in range(10):

p = Process(target=work, args=(i,)) # target传入目标函数，args传入目标函数所需参数

p.start() # 启动进程

通过这样的方式我们就创建了一个简单的进程，需要注意的是，这边的进程是每创建一个，就直接启动该进程。结果如下：

调用第0个进程
调用第1个进程
调用第2个进程
调用第4个进程
调用第5个进程
调用第3个进程
调用第6个进程
调用第7个进程
调用第8个进程
调用第9个进程

调用第0个进程

调用第1个进程

调用第2个进程

调用第4个进程

调用第5个进程

调用第3个进程

调用第6个进程

调用第7个进程

调用第8个进程

调用第9个进程

继承Process类来创建进程

我们也可以通过编写一个自定义的类，通过继承 Process 类的方式，将目标函数写进 run() 方法中，实现多进程的调用。

from multiprocessing import Process

class MyProcess(Process):
    def __init__(self, arg):
        super(MyProcess, self).__init__()
        self.arg = arg
    def run(self):
        # 将目标函数写进run方法中
        print('调用第{}个进程'.format(self.arg))

if __name__ == '__main__':
    # 创建10个进程 
    for i in range(10): 
        p = MyProcess(i)   
        p.start() # 启动进程

from multiprocessing import Process

class MyProcess(Process):

def __init__(self, arg):

super(MyProcess, self).__init__()

self.arg = arg

def run(self):

# 将目标函数写进run方法中

print('调用第{}个进程'.format(self.arg))

if __name__ == '__main__':

# 创建10个进程

for i in range(10):

p = MyProcess(i)

p.start() # 启动进程

结果是一样的。

效率对比

为了比较多进程的效率，我们将其与单进程比较，查看所消耗的时间。

import os
import time
from multiprocessing import Process

def work(num):
    for i in range(num*100000000):
        a = 0 * i
    print('进程pid为{}.'.format(os.getpid()))

if __name__ == '__main__':
    print('父进程pid为{}.'.format(os.getpid()))
    # 单进程
    t0 = time.time()
    work(3)
    work(3)
    t1 = time.time()
    print('单进程所花时间{}'.format(t1-t0))

    # 多进程
    p1 = Process(target=work, args=(3,))
    p2 = Process(target=work, args=(3,))
    t2 = time.time()
    p1.start()  # 子进程开始，调用start的时候就自动调用了多进程类默认的run()方法
    p2.start() 
    p1.join()   # 使主进程阻塞，等待子进程结束后再执行主进程的程序，在这里，主进程剩下的任务就只有计算t3和print时间
    p2.join()
    t3 = time.time()
    print('多进程所花时间{}'.format(t3-t2))

import os

import time

from multiprocessing import Process

def work(num):

for i in range(num*100000000):

a = 0 * i

print('进程pid为{}.'.format(os.getpid()))

if __name__ == '__main__':

print('父进程pid为{}.'.format(os.getpid()))

# 单进程

t0 = time.time()

work(3)

t1 = time.time()

print('单进程所花时间{}'.format(t1-t0))

# 多进程

p1 = Process(target=work, args=(3,))

p2 = Process(target=work, args=(3,))

t2 = time.time()

p1.start() # 子进程开始，调用start的时候就自动调用了多进程类默认的run()方法

p2.start()

p1.join() # 使主进程阻塞，等待子进程结束后再执行主进程的程序，在这里，主进程剩下的任务就只有计算t3和print时间

p2.join()

t3 = time.time()

print('多进程所花时间{}'.format(t3-t2))

上面的代码定义了一个多次数据相乘计算的耗时函数，在运行结束时打印调用此函数的进程ID，首先会计算单进程所花时间，此时的单进程就是主进程。而在多进程计算中，我们创建了两个 Process 类的实例，并指定目标函数 work ，双进程并行执行，执行完成后打印所消耗的时间，它的运算结果如下：

父进程pid为8832.
进程pid为8832.
进程pid为8832.
单进程所花时间18.594270706176758
进程pid为8924.
进程pid为9964.
多进程所花时间10.483423948287964

父进程pid为8832.

进程pid为8832.

单进程所花时间18.594270706176758

进程pid为8924.

进程pid为9964.

多进程所花时间10.483423948287964

可以看出单进程的 pid 号都是一样的，为 8832 。而多进程中，主进程是 8832 ，两个子进程 pid 为 8924 和 9964 。计算所消耗的时间也缩减了将近一半。

注：在Python IDLE中不会显示子进程函数中打印的内容。

join()的作用

在上面的例子中我们加入了 join() 方法等待子进程结束后，再继续执行主进程。如果不加的话，就会出现主进程先运行完毕，然后再运行完子进程的情况。

import os
import time
from multiprocessing import Process

def work(num):
    for i in range(num*100000000):
        a = 0 * i
    print('进程pid为{}.'.format(os.getpid()))

if __name__ == '__main__':
    print('父进程pid为{}.'.format(os.getpid()))
    # 单进程
    t0 = time.time()
    work(3)
    work(3)
    t1 = time.time()
    print('单进程所花时间{}'.format(t1-t0))

    # 多进程
    p1 = Process(target=work, args=(3,))
    p2 = Process(target=work, args=(3,))
    t2 = time.time()
    p1.start()  # 子进程开始，调用start的时候就自动调用了多进程类默认的run()方法
    p2.start() 
    t3 = time.time()
    print('多进程所花时间{}'.format(t3-t2))

import os

import time

from multiprocessing import Process

def work(num):

for i in range(num*100000000):

a = 0 * i

print('进程pid为{}.'.format(os.getpid()))

if __name__ == '__main__':

print('父进程pid为{}.'.format(os.getpid()))

# 单进程

t0 = time.time()

work(3)

t1 = time.time()

print('单进程所花时间{}'.format(t1-t0))

# 多进程

p1 = Process(target=work, args=(3,))

p2 = Process(target=work, args=(3,))

t2 = time.time()

p1.start() # 子进程开始，调用start的时候就自动调用了多进程类默认的run()方法

p2.start()

t3 = time.time()

print('多进程所花时间{}'.format(t3-t2))

我们针对上述的代码，仅仅去掉了 p1.join() 和 p2.join() 两列，就会造成如下结果：

父进程pid为3228.
进程pid为3228.
进程pid为3228.
单进程所花时间25.513729333877563
多进程所花时间0.007950067520141602
进程pid为12120.
进程pid为10360.

父进程pid为3228.

进程pid为3228.

单进程所花时间25.513729333877563

多进程所花时间0.007950067520141602

进程pid为12120.

进程pid为10360.

可以看出主进程最后本来设想打印多进程所花时间的语句会提前执行，并没有等待子进程结束。而主进程早已运行完后，两个子进程才分别输出进程 pid 。

daemon属性

说到这，有人就会问了：有没有当主进程结束，不管子进程结束与否都强制结束的方法，也就是说主进程结束，整个程序就结束了，子进程有可能被强制结束。

这个方法就是在进程 start() 前，设置 daemon 属性。

import os
import time
from multiprocessing import Process

def work(num):
    for i in range(num*100000000):
        a = 0 * i
    print('进程pid为{}.'.format(os.getpid()))

if __name__ == '__main__':
    print('父进程pid为{}.'.format(os.getpid()))
    # 单进程
    t0 = time.time()
    work(3)
    work(3)
    t1 = time.time()
    print('单进程所花时间{}'.format(t1-t0))

    # 多进程
    p1 = Process(target=work, args=(3,))
    p2 = Process(target=work, args=(3,))
    t2 = time.time()
    p1.daemon = True    # 表示主进程结束，程序就完全结束，不管子进程是否运行完毕
    p2.daemon = True
    p1.start()  # 子进程开始，调用start的时候就自动调用了多进程类默认的run()方法
    p2.start() 
    t3 = time.time()
    print('多进程所花时间{}'.format(t3-t2))

import os

import time

from multiprocessing import Process

def work(num):

for i in range(num*100000000):

a = 0 * i

print('进程pid为{}.'.format(os.getpid()))

if __name__ == '__main__':

print('父进程pid为{}.'.format(os.getpid()))

# 单进程

t0 = time.time()

work(3)

t1 = time.time()

print('单进程所花时间{}'.format(t1-t0))

# 多进程

p1 = Process(target=work, args=(3,))

p2 = Process(target=work, args=(3,))

t2 = time.time()

p1.daemon = True # 表示主进程结束，程序就完全结束，不管子进程是否运行完毕

p2.daemon = True

p1.start() # 子进程开始，调用start的时候就自动调用了多进程类默认的run()方法

p2.start()

t3 = time.time()

print('多进程所花时间{}'.format(t3-t2))

得到如下结果：

父进程pid为8052.
进程pid为8052.
进程pid为8052.
单进程所花时间18.618181705474854
多进程所花时间0.006980419158935547

父进程pid为8052.

进程pid为8052.

单进程所花时间18.618181705474854

多进程所花时间0.006980419158935547

可以看出，在主进程打印出多进程所花时间后就已经结束，而此刻的两个子进程被强制终止了。

进程池Pool方法

在使用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机并行操作，可以节约大量的时间。当被操作对象数目不大时，可以直接利用 multiprocessing.Process 动态生成多个进程，但如果是上百个、上千个目标，手动限制进程数量又太繁琐，尤其涉及到进程的销毁。而进程池就发挥出了它的功效。

进程池Pool方法可以提供指定数量的进程供我们调用，当有新的请求提交到Pool中时，如果池未满，就会创建一个新的进程执行该请求；如果池中已经达到最大进程数，该请求就会等待，直到池中有进程结束才会为此请求创建新的进程。进程池中的数量最好等于CPU核心数量。

Pool类的构造方法为：

class multiprocessing.Pool(processes=None, initializer=None, initargs=(), maxtasksperchild=None, context=None)

1	class multiprocessing.Pool(processes=None, initializer=None, initargs=(), maxtasksperchild=None, context=None)

参数说明：

processes：工作进程的数量，如果processes是None，使用os.cpu_count()返回当前cpu的核数；
initializer：如果为None，那么在每个工作进程开始时都会调用initializer方法；
maxtasksperchild：工作进程退出前可以完成的任务数。完成后用一个新的工作进程来替代原进程，以便让闲置的资源得到解放。默认为None，也就是说只要Pool存在工作进程，就会一直存活；
context：用于指定工作进程启动时的上下文。

包含的方法：

apply：同步进程池，主进程会阻塞子函数。阻塞可以理解成串行排队；
apply_async：异步进程池，非阻塞且支持结果返回后进行回调，在没有使用join()方法时，主进程循环运行过程中不等待apply_async的返回结果，在主进程结束后，即使子进程还没返回，整个程序也会退出。返回结果的get()方法是阻塞的，如使用result.get()会阻塞主进程；
map：单个任务会并行运行，会使进程阻塞，直到结果返回。第二个参数为iterable。在实际使用时，只有在整个队列全部就绪后，程序才会运行子进程；
map_async：与map用法一致，是非阻塞的；
close：关闭进程池，组织更多的任务提交到进程池，待任务完成后，工作进程会退出；
terminate：结束工作进程，不再处理未完成的任务；
join：主进程阻塞，等待子进程退出。join()方法要在close()或terminate()之后使用。

怎么取得进程的结果？

阻塞式函数：
Pool.apply() 直接返回结果
Pool.map() 直接返回一个 list
非阻塞式函数：
Pool.apply_async() 和 Pool.map_async() 返回一个 AsyncResult 对象。
AsyncResult 对象具有: get() 函数可以获取结果。
imap() 和 imap_unordered() 则是返回可迭代函数。

注：close()与terminate()的区别在于close()会等待池中的工作进程执行结束后再关闭pool，而terminate()是直接关闭pool。

事实上， apply 方法并不实用，有人也建议将其剔除。我们以 apply_async 示例，这也是用的比较多的方法。

import os
import time
from multiprocessing import Pool

def work(num):
    for i in range(100000000):
        a = 0 * i
    print('{}, 进程pid为{}, id={}.'.format(time.strftime('%H:%M:%S'), os.getpid(), num))

if __name__ == '__main__':
    # 多进程
    t2 = time.time()
    p = Pool(5)
    for i in range(10):
        p.apply_async(func=work, args=(i,))
    p.close()    # 与Process不同，没有start方法，但是需要close
    p.join()   # 不使用join方法阻塞主进程的话，就相当于用了daemon方法，主程序结束整个程序就结束了，不管子进程状态
    t3 = time.time()
    print('多进程所花时间{}'.format(t3-t2))

import os

import time

from multiprocessing import Pool

def work(num):

for i in range(100000000):

a = 0 * i

print('{}, 进程pid为{}, id={}.'.format(time.strftime('%H:%M:%S'), os.getpid(), num))

if __name__ == '__main__':

# 多进程

t2 = time.time()

p = Pool(5)

for i in range(10):

p.apply_async(func=work, args=(i,))

p.close() # 与Process不同，没有start方法，但是需要close

p.join() # 不使用join方法阻塞主进程的话，就相当于用了daemon方法，主程序结束整个程序就结束了，不管子进程状态

t3 = time.time()

print('多进程所花时间{}'.format(t3-t2))

我们在这里创建了一个大小为5的进程池，提交了10个请求任务，结果如下：

16:27:47, 进程pid为12344, id=4.
16:27:47, 进程pid为13972, id=1.
16:27:47, 进程pid为8472, id=0.
16:27:47, 进程pid为5212, id=3.
16:27:47, 进程pid为6364, id=2.
16:27:51, 进程pid为12344, id=5.
16:27:51, 进程pid为13972, id=6.
16:27:51, 进程pid为8472, id=7.
16:27:51, 进程pid为6364, id=9.
16:27:51, 进程pid为5212, id=8.
多进程所花时间8.024072408676147

16:27:47, 进程pid为12344, id=4.

16:27:47, 进程pid为13972, id=1.

16:27:47, 进程pid为8472, id=0.

16:27:47, 进程pid为5212, id=3.

16:27:47, 进程pid为6364, id=2.

16:27:51, 进程pid为12344, id=5.

16:27:51, 进程pid为13972, id=6.

16:27:51, 进程pid为8472, id=7.

16:27:51, 进程pid为6364, id=9.

16:27:51, 进程pid为5212, id=8.

多进程所花时间8.024072408676147

从时间上可以看出，前5个请求是同一时间开始的，而后续的5个请求基本上是在前5个进程都结束后创建的新进程。

进程间数据共享

多个进程间如果需要访问某个共享资源，为了防止共享资源被同时篡改，可以使用锁Lock和队列Queue来实现。可以这样来理解锁Lock：把进程比作一个上厕所的人，当他进入厕所时，需要锁门，此时外面的人看到厕所门上锁了就会在门口排队。也就是说这样保证了在某一时间只能有一个进程访问某个共享资源。

进程同步之Lock

直接来看不加锁的情况，代码如下：

import time
from multiprocessing import Process

def work(name):
    for i in range(5):
        print('{}, {}'.format(time.strftime('%H:%M:%S'), name))
        time.sleep(1)

if __name__ == '__main__':
    p1 = Process(target=work, args=('task1',))
    p2 = Process(target=work, args=('task2',))
    p3 = Process(target=work, args=('task3',))
    p1.start()
    p2.start()
    p3.start()

import time

from multiprocessing import Process

def work(name):

for i in range(5):

print('{}, {}'.format(time.strftime('%H:%M:%S'), name))

time.sleep(1)

if __name__ == '__main__':

p1 = Process(target=work, args=('task1',))

p2 = Process(target=work, args=('task2',))

p3 = Process(target=work, args=('task3',))

p1.start()

p2.start()

p3.start()

上述代码未使用锁，生成了三个子进程，每个进程都打印自己的信息。运行结果如下：

21:00:28, task1
21:00:28, task2
21:00:28, task3
21:00:29, task1
21:00:29, task2
21:00:29, task3
21:00:30, task1
21:00:30, task2
21:00:30, task3
21:00:31, task1
21:00:31, task2
21:00:31, task3
21:00:32, task1
21:00:32, task2
21:00:32, task3

21:00:28, task1

21:00:28, task2

21:00:28, task3

21:00:29, task1

21:00:29, task2

21:00:29, task3

21:00:30, task1

21:00:30, task2

21:00:30, task3

21:00:31, task1

21:00:31, task2

21:00:31, task3

21:00:32, task1

21:00:32, task2

21:00:32, task3

在同一时刻有3个进程都在打印信息，实际应用中，可能会造成信息混乱。现在我们修改一下上面的程序，要求同一时刻只有一个进程在输出信息。

import time
from multiprocessing import Process, Lock

def work(name, lock):
    lock.acquire()
    for i in range(5):
        print('{}, {}'.format(time.strftime('%H:%M:%S'), name))
        time.sleep(1)
    lock.release()

if __name__ == '__main__':
    lock = Lock()
    p1 = Process(target=work, args=('task1', lock))
    p2 = Process(target=work, args=('task2', lock))
    p3 = Process(target=work, args=('task3', lock))
    p1.start()
    p2.start()
    p3.start()

import time

from multiprocessing import Process, Lock

def work(name, lock):

lock.acquire()

for i in range(5):

print('{}, {}'.format(time.strftime('%H:%M:%S'), name))

time.sleep(1)

lock.release()

if __name__ == '__main__':

lock = Lock()

p1 = Process(target=work, args=('task1', lock))

p2 = Process(target=work, args=('task2', lock))

p3 = Process(target=work, args=('task3', lock))

p1.start()

p2.start()

p3.start()

在上面的代码中，我们在每个子进程函数中都加了锁Lock。锁的使用非常简单，首先初始化一个锁的实例 lock = Lock() ，然后在需要独占的代码前加锁，即调用 lock.acquire() ，运行完成后释放锁，即调用 lock.release() ；也可以使用上下文关键字 with 来加锁，如下所示：

def work(name, lock):
    with lock:
        for i in range(5):
            print('{}, {}'.format(time.strftime('%H:%M:%S'), name))
            time.sleep(1)

def work(name, lock):

with lock:

for i in range(5):

print('{}, {}'.format(time.strftime('%H:%M:%S'), name))

time.sleep(1)

最终的输出如下所示：

21:19:59, task1
21:20:00, task1
21:20:01, task1
21:20:02, task1
21:20:03, task1
21:20:04, task2
21:20:05, task2
21:20:06, task2
21:20:07, task2
21:20:08, task2
21:20:09, task3
21:20:10, task3
21:20:11, task3
21:20:12, task3
21:20:13, task3

21:19:59, task1

21:20:00, task1

21:20:01, task1

21:20:02, task1

21:20:03, task1

21:20:04, task2

21:20:05, task2

21:20:06, task2

21:20:07, task2

21:20:08, task2

21:20:09, task3

21:20:10, task3

21:20:11, task3

21:20:12, task3

21:20:13, task3

从输出结果可以看出，同一时刻仅有一个进程在输出信息。

进程优先级队列Queue

Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。

Queue队列是先进先出的，它有两个方法：

put方法：将数据插入队列中。有两个可选参数blocked和timeout。如果blocked为True（默认值），并且timeout为正值，则该方法阻塞timeout时间，直到该队列有剩余的空间。如果超时，则会抛出Queue.Full异常。如果blocked为False，但该Queue已满，则会立即抛出Queue.Full异常。
get方法：从队列中读取并删除一个元素。同样有两个参数blocked和timeout。如果blocked为True（默认值），timeout为正值，在等待时间内没有取到任何元素，则会抛出Queue.Empty异常。如果blocked为False，Queue有一个值可用，立即返回该值，否则Queue为空，立即抛出Queue.Empty异常。

Queue常常用在生产者-消费者模式中，这种模式下，生产者的产量和消费者的消耗量在一定的时间内是不一致的。有时候产量多而消耗量少，此时需要让生产者休息一下，等产量消耗一部分后继续生产；有时候消耗量过多，而产量较小的情况下，就需要消费者暂时休息一下，等产量上来后在继续消耗。这样一种模式的应用非常广泛。Queue在其中起到了缓存的作用。

《手把手教你实现Python多进程多线程》

使用多进程实现生产者-消费者模式，示例代码如下：

import time
from multiprocessing import Process, Queue

def Producer(queue):
    num = 1
    while True:
        queue.put('iphone {}'.format(num))
        print('{}，生产者生产了[iphone {}]'.format(time.strftime('%H:%M:%S'), num))
        num += 1
        time.sleep(1)

def Consumer(queue):
    while True:
        print('{}，消费者购买了[{}]'.format(time.strftime('%H:%M:%S'), queue.get()))
        time.sleep(5)

if __name__ == '__main__':
    queue = Queue(maxsize=5)
    p = Process(target=Producer, args=(queue,))
    c = Process(target=Consumer, args=(queue,))
    c.start()
    p.start()
    c.join()
    p.join()

import time

from multiprocessing import Process, Queue

def Producer(queue):

num = 1

while True:

queue.put('iphone {}'.format(num))

print('{}，生产者生产了[iphone {}]'.format(time.strftime('%H:%M:%S'), num))

num += 1

time.sleep(1)

def Consumer(queue):

while True:

print('{}，消费者购买了[{}]'.format(time.strftime('%H:%M:%S'), queue.get()))

time.sleep(5)

if __name__ == '__main__':

queue = Queue(maxsize=5)

p = Process(target=Producer, args=(queue,))

c = Process(target=Consumer, args=(queue,))

c.start()

p.start()

c.join()

p.join()

上述代码定义了生产者和消费者，队列的最大容量为5，生产者会不断生产iphone放入队列中，消费者不断从队列中购买iphone，队列满时，生产者等待，当队列空时，消费者等待。最终Queue让生产者和消费者有条不紊的一直进程下去。运行的结果如下所示：

22:22:56，生产者生产了[iphone 1]
22:22:56，消费者购买了[iphone 1]
22:22:57，生产者生产了[iphone 2]
22:22:58，生产者生产了[iphone 3]
22:22:59，生产者生产了[iphone 4]
22:23:00，生产者生产了[iphone 5]
22:23:01，生产者生产了[iphone 6]
22:23:01，消费者购买了[iphone 2]
22:23:02，生产者生产了[iphone 7]
22:23:06，消费者购买了[iphone 3]
22:23:06，生产者生产了[iphone 8]
22:23:11，消费者购买了[iphone 4]
22:23:11，生产者生产了[iphone 9]
22:23:16，消费者购买了[iphone 5]
22:23:16，生产者生产了[iphone 10]
...

22:22:56，生产者生产了[iphone 1]

22:22:56，消费者购买了[iphone 1]

22:22:57，生产者生产了[iphone 2]

22:22:58，生产者生产了[iphone 3]

22:22:59，生产者生产了[iphone 4]

22:23:00，生产者生产了[iphone 5]

22:23:01，生产者生产了[iphone 6]

22:23:01，消费者购买了[iphone 2]

22:23:02，生产者生产了[iphone 7]

22:23:06，消费者购买了[iphone 3]

22:23:06，生产者生产了[iphone 8]

22:23:11，消费者购买了[iphone 4]

22:23:11，生产者生产了[iphone 9]

22:23:16，消费者购买了[iphone 5]

22:23:16，生产者生产了[iphone 10]

...

从结果也能看出，生产者前期速度很快，在队列满时，等待消费者消耗队列资源后，才会进一步生产。

多线程创建方法Thread类

在掌握了多进程方法后，多线程方法就很容易理解了。多线程与多进程的调用具有相似性，Python内置的 threading 模块提供了创建多线程的方法。

Thread 类的构造方法：

class threading.Thread(group=None, target=None, name=None, args=(), kwargs={})

1	class threading.Thread(group=None, target=None, name=None, args=(), kwargs={})

参数说明：

group表示线程组，目前还没有实现，必须是None；
target表示要执行的函数；
name表示线程名；
args/kwargs表示要传入线程函数的参数。

最基本的 Thread 类同样具有两种创建方式：

方法一：直接创建一个 Thread 类的实例，并传入目标函数。

from threading import Thread
 
def work(i):
    print('调用第{}个线程'.format(i))
 
if __name__ == '__main__':
    # 创建10个线程
    for i in range(10):
        t = Thread(target=work, args=(i,))  # target传入目标函数，args传入目标函数所需参数
        t.start()  # 启动线程

from threading import Thread

def work(i):

print('调用第{}个线程'.format(i))

if __name__ == '__main__':

# 创建10个线程

for i in range(10):

t = Thread(target=work, args=(i,)) # target传入目标函数，args传入目标函数所需参数

t.start() # 启动线程

输入如下结果：

调用第0个线程
调用第1个线程
调用第2个线程
调用第3个线程
调用第4个线程
调用第5个线程
调用第6个线程
调用第7个线程
调用第8个线程
调用第9个线程

调用第0个线程

调用第1个线程

调用第2个线程

调用第3个线程

调用第4个线程

调用第5个线程

调用第6个线程

调用第7个线程

调用第8个线程

调用第9个线程

对比多进程的 Process 方法，可以发现它们是类似的。

方法二：自定义一个类并继承 Thread 类，重写其 __init__() 方法和 run() 方法。

from threading import Thread

class MyThread(Thread):
    def __init__(self, arg):
        super(MyThread, self).__init__()
        self.arg = arg
    def run(self):
        # 将目标函数写进run方法中
        print('调用第{}个线程'.format(self.arg))

if __name__ == '__main__':
    # 创建10个线程 
    for i in range(10): 
        p = MyThread(i)   
        p.start() # 启动线程

from threading import Thread

class MyThread(Thread):

def __init__(self, arg):

super(MyThread, self).__init__()

self.arg = arg

def run(self):

# 将目标函数写进run方法中

print('调用第{}个线程'.format(self.arg))

if __name__ == '__main__':

# 创建10个线程

for i in range(10):

p = MyThread(i)

p.start() # 启动线程

多线程之线程池pool

在面向对象编程中，创建和销毁对象比较费时，线程池能够很好的解决这个问题。将任务添加到线程池中，线程池会自动指定一个空闲的线程去执行任务，当超过线程池的最大线程数时，任务需要等待有新的空闲线程后才会被执行。

ThreadPool 方法跟多进程类似，只是导入方式不一样，在 multiprocessing.pool.ThreadPool 调用线程池：

import os
import time
from multiprocessing.pool import ThreadPool

def work():
    for i in range(100000000):
        a = 0 * i
    
if __name__ == '__main__':
    # 多进程
    t2 = time.time()
    t = ThreadPool(5)
    for i in range(10):
        t.apply_async(func=work)
    t.close()    
    t.join()  
    t3 = time.time()
    print('多线程所花时间{}'.format(t3-t2))

import os

import time

from multiprocessing.pool import ThreadPool

def work():

for i in range(100000000):

a = 0 * i

if __name__ == '__main__':

# 多进程

t2 = time.time()

t = ThreadPool(5)

for i in range(10):

t.apply_async(func=work)

t.close()

t.join()

t3 = time.time()

print('多线程所花时间{}'.format(t3-t2))

结果为：

多线程所花时间32.743911027908325

1	多线程所花时间32.743911027908325

这个结果明显比多进程的8秒时间更长，这是因为这是一个需要调用计算资源的任务，多进程更有优势。

多线程同步之Lock（互斥锁）

多线程也有锁Lock，这是因为当多个线程一起对某个数据进行修改，可能出现不可预料的结果，这个时候就需要使用锁Lock，还是那个“上厕所”的比喻。我们在这抛砖引玉，举个简单的例子：

from threading import Thread

def work(n):
    global num
    for i in range(1000000):
        num += n
        num -= n

if __name__ == '__main__':
    num = 0
    # 创建3个线程 
    t1 = Thread(target=work, args=(3,))
    t2 = Thread(target=work, args=(14,))
    t3 = Thread(target=work, args=(25,))
    t1.start()
    t2.start()
    t3.start()
    t1.join()
    t2.join()
    t3.join()
    print(num)

from threading import Thread

def work(n):

global num

for i in range(1000000):

num += n

num -= n

if __name__ == '__main__':

num = 0

# 创建3个线程

t1 = Thread(target=work, args=(3,))

t2 = Thread(target=work, args=(14,))

t3 = Thread(target=work, args=(25,))

t1.start()

t2.start()

t3.start()

t1.join()

t2.join()

t3.join()

print(num)

这里的 work 函数只是单纯的加减同一个数，最终 num 的值应该是为0的。我们可以看运行结果：

-25

-25

每次执行发现结果都不一样，这是怎么回事呢？原来我们在这里创建了三个线程，当线程 t1 在执行 num += n 时，有可能其它线程也正在执行 num += n 或 num -= n ，因此导致 num 值不是加了又减同一个数。使得最后的 num 值不为0。为了让一个线程在访问数据时，其它线程在等待，就需要用到锁Lock。看下面的代码：

from threading import Thread, Lock

def work(n):
    global num
    lock.acquire()
    for i in range(1000000):
        num += n
        num -= n
    lock.release()

if __name__ == '__main__':
    num = 0
    lock = Lock()
    # 创建3个线程 
    t1 = Thread(target=work, args=(3,))
    t2 = Thread(target=work, args=(14,))
    t3 = Thread(target=work, args=(25,))
    t1.start()
    t2.start()
    t3.start()
    t1.join()
    t2.join()
    t3.join()
    print(num)

from threading import Thread, Lock

def work(n):

global num

lock.acquire()

for i in range(1000000):

num += n

num -= n

lock.release()

if __name__ == '__main__':

num = 0

lock = Lock()

# 创建3个线程

t1 = Thread(target=work, args=(3,))

t2 = Thread(target=work, args=(14,))

t3 = Thread(target=work, args=(25,))

t1.start()

t2.start()

t3.start()

t1.join()

t2.join()

t3.join()

print(num)

加锁的方式在多进程也介绍过，最终无论怎么运行，结果都只能为0。

线程优先级队列Queue

与多进程一样，队列相当于一个缓冲区，实现线程之间的通信和数据共享的安全。同样以生产者-消费者模式为例：

import time
from queue import Queue
from threading import Thread

def Producer(queue):
    num = 1
    while True:
        queue.put('iphone {}'.format(num))
        print('{}，生产者生产了[iphone {}]'.format(time.strftime('%H:%M:%S'), num))
        num += 1
        time.sleep(1)

def Consumer(queue):
    while True:
        print('{}，消费者购买了[{}]'.format(time.strftime('%H:%M:%S'), queue.get()))
        time.sleep(5)

if __name__ == '__main__':
    queue = Queue(maxsize=5)
    p = Thread(target=Producer, args=(queue,))
    c = Thread(target=Consumer, args=(queue,))
    c.start()
    p.start()
    c.join() # 没有主线程内容，可省略
    p.join()

import time

from queue import Queue

from threading import Thread

def Producer(queue):

num = 1

while True:

queue.put('iphone {}'.format(num))

print('{}，生产者生产了[iphone {}]'.format(time.strftime('%H:%M:%S'), num))

num += 1

time.sleep(1)

def Consumer(queue):

while True:

print('{}，消费者购买了[{}]'.format(time.strftime('%H:%M:%S'), queue.get()))

time.sleep(5)

if __name__ == '__main__':

queue = Queue(maxsize=5)

p = Thread(target=Producer, args=(queue,))

c = Thread(target=Consumer, args=(queue,))

c.start()

p.start()

c.join() # 没有主线程内容，可省略

p.join()

Python的 queue 模块提供了队列 Queue 方法，详细信息在多进程中也介绍了，在此不再赘述。

12:25:20，生产者生产了[iphone 1]
12:25:20，消费者购买了[iphone 1]
12:25:21，生产者生产了[iphone 2]
12:25:22，生产者生产了[iphone 3]
12:25:23，生产者生产了[iphone 4]
12:25:24，生产者生产了[iphone 5]
12:25:25，消费者购买了[iphone 2]
12:25:25，生产者生产了[iphone 6]
12:25:26，生产者生产了[iphone 7]
12:25:30，消费者购买了[iphone 3]
12:25:30，生产者生产了[iphone 8]
12:25:35，消费者购买了[iphone 4]
12:25:35，生产者生产了[iphone 9]
12:25:40，消费者购买了[iphone 5]
...

12:25:20，生产者生产了[iphone 1]

12:25:20，消费者购买了[iphone 1]

12:25:21，生产者生产了[iphone 2]

12:25:22，生产者生产了[iphone 3]

12:25:23，生产者生产了[iphone 4]

12:25:24，生产者生产了[iphone 5]

12:25:25，消费者购买了[iphone 2]

12:25:25，生产者生产了[iphone 6]

12:25:26，生产者生产了[iphone 7]

12:25:30，消费者购买了[iphone 3]

12:25:30，生产者生产了[iphone 8]

12:25:35，消费者购买了[iphone 4]

12:25:35，生产者生产了[iphone 9]

12:25:40，消费者购买了[iphone 5]

...

多进程vs多线程

在最后一部分，我们来真实比较下多进程和多线程在计算密集型任务和I/O密集型任务中的表现，切实感受下到底哪种方式做并行计算更优。

计算密集型-多进程

import os
import time
from multiprocessing import Process

# 计算密集型
def work():
    n = 0
    for i in range(100000000):
        n *= i

if __name__ == '__main__':
    alist = []
    print('本机为{}核CPU'.format(os.cpu_count()))
    start = time.time()
    for i in range(4):
        p = Process(target=work)
        alist.append(p)
        p.start()
    for p in alist:
        p.join()
    stop = time.time()
    print('计算密集型任务所耗时{}'.format(stop - start))

import os

import time

from multiprocessing import Process

# 计算密集型

def work():

n = 0

for i in range(100000000):

n *= i

if __name__ == '__main__':

alist = []

print('本机为{}核CPU'.format(os.cpu_count()))

start = time.time()

for i in range(4):

p = Process(target=work)

alist.append(p)

p.start()

for p in alist:

p.join()

stop = time.time()

print('计算密集型任务所耗时{}'.format(stop - start))

结果为：

本机为12核CPU
计算密集型任务所耗时4.171813011169434

1 2	本机为12核CPU 计算密集型任务所耗时4.171813011169434

计算密集型-多线程

import os
import time
from threading import Thread

# 计算密集型
def work():
    n = 0
    for i in range(100000000):
        n *= i

if __name__ == '__main__':
    alist = []
    print('本机为{}核CPU'.format(os.cpu_count()))
    start = time.time()
    for i in range(4):
        t = Thread(target=work)
        alist.append(t)
        t.start()
    for t in alist:
        t.join()
    stop = time.time()
    print('计算密集型任务所耗时{}'.format(stop - start))

import os

import time

from threading import Thread

# 计算密集型

def work():

n = 0

for i in range(100000000):

n *= i

if __name__ == '__main__':

alist = []

print('本机为{}核CPU'.format(os.cpu_count()))

start = time.time()

for i in range(4):

t = Thread(target=work)

alist.append(t)

t.start()

for t in alist:

t.join()

stop = time.time()

print('计算密集型任务所耗时{}'.format(stop - start))

结果为：

本机为12核CPU
计算密集型任务所耗时14.696670532226562

1 2	本机为12核CPU 计算密集型任务所耗时14.696670532226562

明显可以看出，多进程对计算密集型更有利。

I/O密集型-多进程

import os
import time
from multiprocessing import Process

# I/O密集型
def work():
    time.sleep(2)
    print('===>', file=open('tmp.txt', 'w'))

if __name__ == '__main__':
    alist = []
    print('本机为{}核CPU'.format(os.cpu_count()))
    start = time.time()
    for i in range(400):
        p = Process(target=work)
        alist.append(p)
        p.start()
    for p in alist:
        p.join()
    stop = time.time()
    print('I/O密集型任务所耗时{}'.format(stop - start))

import os

import time

from multiprocessing import Process

# I/O密集型

def work():

time.sleep(2)

print('===>', file=open('tmp.txt', 'w'))

if __name__ == '__main__':

alist = []

print('本机为{}核CPU'.format(os.cpu_count()))

start = time.time()

for i in range(400):

p = Process(target=work)

alist.append(p)

p.start()

for p in alist:

p.join()

stop = time.time()

print('I/O密集型任务所耗时{}'.format(stop - start))

结果为：

本机为12核CPU
I/O密集型任务所耗时5.8300940990448

1 2	本机为12核CPU I/O密集型任务所耗时5.8300940990448

I/O密集型-多线程

import os
import time
from threading import Thread

# I/O密集型
def work():
    time.sleep(2)
    print('===>', file=open('tmp.txt', 'w'))

if __name__ == '__main__':
    alist = []
    print('本机为{}核CPU'.format(os.cpu_count()))
    start = time.time()
    for i in range(400):
        t = Thread(target=work)
        alist.append(t)
        t.start()
    for t in alist:
        t.join()
    stop = time.time()
    print('I/O密集型任务所耗时{}'.format(stop - start))

import os

import time

from threading import Thread

# I/O密集型

def work():

time.sleep(2)

print('===>', file=open('tmp.txt', 'w'))

if __name__ == '__main__':

alist = []

print('本机为{}核CPU'.format(os.cpu_count()))

start = time.time()

for i in range(400):

t = Thread(target=work)

alist.append(t)

t.start()

for t in alist:

t.join()

stop = time.time()

print('I/O密集型任务所耗时{}'.format(stop - start))

结果为：

本机为12核CPU
I/O密集型任务所耗时2.0723369121551514

1 2	本机为12核CPU I/O密集型任务所耗时2.0723369121551514

上述可以看出，对于I/O密集型任务，多线程速度更快。

总结

本文对多线程多进程常用方式进行了详细的总结。在Python程序设计中，要根据任务本身的特点选择合适的并行方式，从而大大加快程序的运行速度。

lucky说道：

2020年12月11日下午1:57

点赞，不错

回复

简介

多进程、多线程如何选择

多进程创建方法Process类

直接创建Process类实例

继承Process类来创建进程

效率对比

join()的作用

daemon属性

进程池Pool方法

怎么取得进程的结果？

进程间数据共享

进程同步之Lock

进程优先级队列Queue

多线程创建方法Thread类

多线程之线程池pool

多线程同步之Lock（互斥锁）

线程优先级队列Queue

多进程vs多线程

计算密集型-多进程

计算密集型-多线程

I/O密集型-多进程

I/O密集型-多线程

总结

相关文章

发表评论 取消回复

发表评论取消回复