赞
踩
目录
(1)Python的多进程编程与multiprocess模块
(2)利用multiprocess模块的Pool类创建多进程
线程:是程序执行流的最小单元,是系统独立调度和分配CPU(独立运行)的基本单位。
进程:是资源分配的基本单位。一个进程包括多个线程。
一个应用程序至少包括1个进程,而1个进程包括1个或多个线程,线程的尺度更小。
每个进程在执行过程中拥有独立的内存单元,而一个线程的多个线程在执行过程中共享内存。
(1)线程与资源分配无关,它属于某一个进程,并与进程内的其他线程一起共享进程的资源。
(2)每个进程都有自己一套独立的资源(数据),供其内的所有线程共享。
(3)不论是大小,开销线程要更“轻量级”。
(4)一个进程内的线程通信比进程之间的通信更快速,有效。(因为共享变量)。
多线程:同一时刻执行多个线程。如,用浏览器一边下载,一边听歌,一边看视频,一边看网页等。
多进程:同时执行多个程序。如,同时运行YY,QQ,以及各种浏览器。
python的多进程编程主要依靠multiprocess模块,我们先对比两段代码,看看多进程编程的优势。我们模拟了一个非常耗时的任务,计算8的20次方,为了使这个任务显得更耗时,我们还让它sleep 2秒。第一段代码是单进程计算(代码如下所示),我们按顺序执行代码,重复计算2次,并打印出总共耗时。
串行执行:
- # -*- coding: UTF-8 -*-
- """
- # 计算8的20次方
- """
- import time
- import os
-
-
- def long_time_task():
- print('当前进程: {}'.format(os.getpid()))
- time.sleep(2)
- print("结果: {}".format(8 ** 20))
-
-
- if __name__ == "__main__":
- print('当前母进程: {}'.format(os.getpid()))
- start = time.time()
- for i in range(2):
- long_time_task()
-
- end = time.time()
- print("用时{}秒".format((end-start)))
运行结果:耗时大概4s多,始终只用一个进程。
- 当前母进程: 96680
- 当前进程: 96680
- 结果: 1152921504606846976
- 当前进程: 96680
- 结果: 1152921504606846976
- 用时4.005445718765259秒
接下来我们来看看multiprocess多进程计算代码。
- # -*- coding: UTF-8 -*-
- """
- # 计算8的20次方
- """
- import time
- import os
-
-
- from multiprocessing import Process
- import os
- import time
-
-
- def long_time_task(i):
- print('子进程: {} - 任务{}'.format(os.getpid(), i))
- time.sleep(2)
- print("结果: {}".format(8 ** 20))
-
-
- if __name__=='__main__':
- print('当前母进程: {}'.format(os.getpid()))
- start = time.time()
- p1 = Process(target=long_time_task, args=(1,))
- p2 = Process(target=long_time_task, args=(2,))
- print('等待所有子进程完成。')
- p1.start()
- p2.start()
- p1.join()
- p2.join()
- end = time.time()
- print("总共用时{}秒".format((end - start)))
运行结果:耗时2s多,时间减少了一半以上,会创建2个进程。
- 当前母进程: 96940
- 等待所有子进程完成。
- 子进程: 96942 - 任务1
- 子进程: 96943 - 任务2
- 结果: 1152921504606846976
- 结果: 1152921504606846976
- 总共用时2.1094207763671875秒
Ps:
很多时候系统都需要创建多个进程以提高CPU的利用率,当数量较少时,可以手动生成一个个Process实例。当进程数量很多时,或许可以循环,但是这需要程序员手动管理系统中并发进程的数量,有时候会很麻烦。这时进程池Pool就可以发挥其功效了。可以通过传递参数限制并非进程的数量,默认值为CPU的核数。
Pool类可以提供指定数量的进程供用户调用,当有新的请求提交到Pool中时,如果进程池还没有满,就会创建一个新的进程来执行请求。如果池满,请求就会告知先等待,直到池中有进程结束,才会创建新的进程来执行这些请求。
下面介绍一下multiprocessing 模块下的Pool类的几个方法:
函数原型:apply_async(func[, args=()[, kwds={}[, callback=None]]])
其作用是向进程池提交需要执行的函数及参数, 各个进程采用非阻塞(异步)的调用方式,即每个子进程只管运行自己的,不管其它进程是否已经完成。这是默认方式。
函数原型:map(func, iterable[, chunksize=None])
Pool类中的map方法,与内置的map函数用法行为基本一致,它会使进程阻塞直到结果返回。 注意:虽然第二个参数是一个迭代器,但在实际使用中,必须在整个队列都就绪后,程序才会运行子进程。
函数原型:map_async(func, iterable[, chunksize[, callback]])
与map用法一致,但是它是非阻塞的。其有关事项见apply_async。
关闭进程池(pool),使其不在接受新的任务。
结束工作进程,不在处理未处理的任务。
主进程阻塞等待子进程的退出, join方法要在close或terminate之后使用。
下例是一个简单的multiprocessing.Pool类的示例。因为我的CPU是8核的,一次最多可以同时运行8个进程,所以我开启了一个容量为8的进程池。8个进程需要计算9次,你可以想象8个进程并行8次计算任务后,还剩一次计算任务(任务8)没有完成,系统会等待8个进程完成后重新安排一个进程来计算。
- # -*- coding: UTF-8 -*-
- """
- # 计算8的20次方
- """
- from multiprocessing import Pool, cpu_count
- import os
- import time
-
-
- def long_time_task(i):
- print('子进程: {} - 任务{}'.format(os.getpid(), i))
- time.sleep(2)
- print("结果: {}".format(8 ** 20))
-
-
- if __name__=='__main__':
- print("CPU内核数:{}".format(cpu_count()))
- print('当前母进程: {}'.format(os.getpid()))
- start = time.time()
- p = Pool(8)
- for i in range(9):
- p.apply_async(long_time_task, args=(i,))
- print('等待所有子进程完成。')
- p.close()
- p.join()
- end = time.time()
- print("总共用时{}秒".format((end - start)))
输出结果:
CPU内核数:8 当前母进程: 98225 等待所有子进程完成。 子进程: 98229 - 任务0 子进程: 98228 - 任务1 子进程: 98227 - 任务2 子进程: 98232 - 任务3 子进程: 98230 - 任务4 子进程: 98231 - 任务5 子进程: 98234 - 任务6 子进程: 98233 - 任务7 结果: 1152921504606846976 子进程: 98229 - 任务8 结果: 1152921504606846976 结果: 1152921504606846976 结果: 1152921504606846976 结果: 1152921504606846976 结果: 1152921504606846976 结果: 1152921504606846976 结果: 1152921504606846976 结果: 1152921504606846976 总共用时4.225358963012695秒
Ps:
相信大家都知道python解释器中存在GIL(全局解释器锁), 它的作用就是保证同一时刻只有一个线程可以执行代码。由于GIL的存在,很多人认为python中的多线程其实并不是真正的多线程,如果想要充分地使用多核CPU的资源,在python中大部分情况需要使用多进程。然而这并意味着python多线程编程没有意义哦,请继续阅读下文。
通常,进程之间是相互独立的,每个进程都有独立的内存。通过共享内存(nmap模块),进程之间可以共享对象,使多个进程可以访问同一个变量(地址相同,变量名可能不同)。多进程共享资源必然会导致进程间相互竞争,所以应该尽最大可能防止使用共享状态。还有一种方式就是使用队列queue来实现不同进程间的通信或数据共享,这一点和多线程编程类似。
下例这段代码中中创建了2个独立进程,一个负责写(pw), 一个负责读(pr), 实现了共享一个队列queue。
- # -*- coding: UTF-8 -*-
- """
- # 计算8的20次方
- """
- from multiprocessing import Process, Queue
- import os, time, random
-
-
- # 写数据进程执行的代码:
- def write(q):
- print('Process to write: {}'.format(os.getpid()))
- for value in ['A', 'B', 'C']:
- print('Put %s to queue...' % value)
- q.put(value)
- time.sleep(random.random())
-
-
- # 读数据进程执行的代码:
- def read(q):
- print('Process to read:{}'.format(os.getpid()))
- while True:
- value = q.get(True)
- print('Get %s from queue.' % value)
-
-
- if __name__=='__main__':
- # 父进程创建Queue,并传给各个子进程:
- q = Queue()
- pw = Process(target=write, args=(q,))
- pr = Process(target=read, args=(q,))
- # 启动子进程pw,写入:
- pw.start()
- # 启动子进程pr,读取:
- pr.start()
- # 等待pw结束:
- pw.join()
- # pr进程里是死循环,无法等待其结束,只能强行终止:
- pr.terminate()
输出结果:
- Process to read:98659
- Process to write: 98658
- Put A to queue...
- Get A from queue.
- Put B to queue...
- Get B from queue.
- Put C to queue...
- Get C from queue.
python 3中的多进程编程主要依靠threading模块。创建新线程与创建新进程的方法非常类似。threading.Thread方法可以接收两个参数, 第一个是target,一般指向函数名,第二个时args,需要向函数传递的参数。对于创建的新线程,调用start()方法即可让其开始。我们还可以使用current_thread().name打印出当前线程的名字。 下例中我们使用多线程技术重构之前的计算代码。
- # -*- coding: UTF-8 -*-
- """
- # 计算8的20次方
- """
- import threading
- import time
-
-
- def long_time_task(i):
- print('当前子线程: {} 任务{}'.format(threading.current_thread().name, i))
- time.sleep(2)
- print("结果: {}".format(8 ** 20))
-
-
- if __name__=='__main__':
- start = time.time()
- print('这是主线程:{}'.format(threading.current_thread().name))
- thread_list = []
- for i in range(1, 3):
- t = threading.Thread(target=long_time_task, args=(i, ))
- thread_list.append(t)
-
- for t in thread_list:
- t.start()
-
- for t in thread_list:
- t.join()
-
- end = time.time()
- print("总共用时{}秒".format((end - start)))
输出结果:
- 这是主线程:MainThread
- 当前子线程: Thread-1 任务1
- 当前子线程: Thread-2 任务2
- 结果: 1152921504606846976结果: 1152921504606846976
-
- 总共用时2.0044820308685303秒
当我们设置多线程时,主线程会创建多个子线程,在python中,默认情况下主线程和子线程独立运行互不干涉。如果希望让主线程等待子线程实现线程的同步,我们需要使用join()方法。如果我们希望一个主线程结束时不再执行子线程,我们应该怎么办呢? 我们可以使set.setDaemon(True),代码如下所示。
- # -*- coding: UTF-8 -*-
- """
- # 计算8的20次方
- """
- import threading
- import time
-
-
- def long_time_task():
- print('当子线程: {}'.format(threading.current_thread().name))
- time.sleep(2)
- print("结果: {}".format(8 ** 20))
-
-
- if __name__=='__main__':
- start = time.time()
- print('这是主线程:{}'.format(threading.current_thread().name))
- for i in range(5):
- t = threading.Thread(target=long_time_task, args=())
- t.setDaemon(True)
- t.start()
-
- end = time.time()
- print("总共用时{}秒".format((end - start)))
运行结果:可以看出,主线程结束时并没有继续执行子线程。
- 这是主线程:MainThread
- 当子线程: Thread-1
- 当子线程: Thread-2
- 当子线程: Thread-3
- 当子线程: Thread-4
- 当子线程: Thread-5
- 总共用时0.0006351470947265625秒
除了使用Thread()方法创建新的线程外,我们还可以通过继承Thread类重写run方法创建新的线程,这种方法更灵活。下例中我们自定义的类为MyThread, 随后我们通过该类的实例化创建了2个子线程。
- #-*- encoding:utf-8 -*-
- import threading
- import time
-
-
- def long_time_task(i):
- time.sleep(2)
- return 8**20
-
-
- class MyThread(threading.Thread):
- def __init__(self, func, args , name='', ):
- threading.Thread.__init__(self)
- self.func = func
- self.args = args
- self.name = name
- self.result = None
-
- def run(self):
- print('开始子进程{}'.format(self.name))
- self.result = self.func(self.args[0],)
- print("结果: {}".format(self.result))
- print('结束子进程{}'.format(self.name))
-
-
- if __name__=='__main__':
- start = time.time()
- threads = []
- for i in range(1, 3):
- t = MyThread(long_time_task, (i,), str(i))
- threads.append(t)
-
- for t in threads:
- t.start()
- for t in threads:
- t.join()
-
- end = time.time()
- print("总共用时{}秒".format((end - start)))
输出结果:
- 开始子进程1开始子进程2
-
- 结果: 1152921504606846976
- 结束子进程2
- 结果: 1152921504606846976
- 结束子进程1
- 总共用时2.0026979446411133秒
一个进程所含的不同线程间共享内存,这就意味着任何一个变量都可以被任何一个线程修改,因此线程之间共享数据最大的危险在于多个线程同时改一个变量,把内容给改乱了。如果不同线程间有共享的变量,其中一个方法就是在修改前给其上一把锁lock,确保一次只有一个线程能修改它。threading.lock()方法可以轻易实现对一个共享变量的锁定,修改完后release供其它线程使用。比如下例中账户余额balance是一个共享变量,使用lock可以使其不被改乱。
- # -*- coding: UTF-8 -*-
- """
- # Demo
- """
- # -*- coding: utf-8 -*
-
- import threading
-
-
- class Account:
- def __init__(self):
- self.balance = 0
-
- def add(self, lock):
- # 获得锁
- lock.acquire()
- for i in range(0, 100000):
- self.balance += 1
- # 释放锁
- lock.release()
-
- def delete(self, lock):
- # 获得锁
- lock.acquire()
- for i in range(0, 100000):
- self.balance -= 1
- # 释放锁
- lock.release()
-
-
- if __name__ == "__main__":
- account = Account()
- lock = threading.Lock()
- # 创建线程
- thread_add = threading.Thread(target=account.add, args=(lock,), name='Add')
- thread_delete = threading.Thread(target=account.delete, args=(lock,), name='Delete')
-
- # 启动线程
- thread_add.start()
- thread_delete.start()
-
- # 等待线程结束
- thread_add.join()
- thread_delete.join()
-
- print('The final balance is: {}'.format(account.balance))
输出结果:
The final balance is: 0
另一种实现不同线程间数据共享的方法就是使用消息队列queue。不像列表,queue是线程安全的,可以放心使用,见下文。
下例中创建了两个线程,一个负责生成,一个负责消费,所生成的产品存放在queue里,实现了不同线程间沟通。
- from queue import Queue
- import random, threading, time
-
-
- # 生产者类
- class Producer(threading.Thread):
- def __init__(self, name, queue):
- threading.Thread.__init__(self, name=name)
- self.queue = queue
-
- def run(self):
- for i in range(1, 5):
- print("{} is producing {} to the queue!".format(self.getName(), i))
- self.queue.put(i)
- time.sleep(random.randrange(10) / 5)
- print("%s finished!" % self.getName())
-
-
- # 消费者类
- class Consumer(threading.Thread):
- def __init__(self, name, queue):
- threading.Thread.__init__(self, name=name)
- self.queue = queue
-
- def run(self):
- for i in range(1, 5):
- val = self.queue.get()
- print("{} is consuming {} in the queue.".format(self.getName(), val))
- time.sleep(random.randrange(10))
- print("%s finished!" % self.getName())
-
-
- def main():
- queue = Queue()
- producer = Producer('Producer', queue)
- consumer = Consumer('Consumer', queue)
-
- producer.start()
- consumer.start()
-
- producer.join()
- consumer.join()
- print('All threads finished!')
-
-
- if __name__ == '__main__':
- main()
队列queue的put方法可以将一个对象obj放入队列中。如果队列已满,此方法将阻塞至队列有空间可用为止。queue的get方法一次返回队列中的一个成员。如果队列为空,此方法将阻塞至队列中有成员可用为止。queue同时还自带emtpy(), full()等方法来判断一个队列是否为空或已满,但是这些方法并不可靠,因为多线程和多进程,在返回结果和使用结果之间,队列中可能添加/删除了成员。
- Producer is producing 1 to the queue!
- Consumer is consuming 1 in the queue.
- Producer is producing 2 to the queue!
- Producer is producing 3 to the queue!
- Producer is producing 4 to the queue!
- Producer finished!
- Consumer is consuming 2 in the queue.
- Consumer is consuming 3 in the queue.
- Consumer is consuming 4 in the queue.
- Consumer finished!
- All threads finished!
由于GIL的存在,很多人认为Python多进程编程更快,针对多核CPU,理论上来说也是采用多进程更能有效利用资源。但这不是说明多线程就没意义了,还是得根据实际场景来看。
好像还是很抽象,该如何理解呢?
对于IO密集型操作,大部分消耗时间其实是等待时间,在等待中CPU是不需要工作的,那么在此期间提供多个CPU资源也是利用不上的,python碰到等待会释放GIL供新的线程使用,实现了线程间的切换。相反对于CPU密集型代码,多个CPU干活肯定比一个CPU快很多。
参考:
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。