问:线程学完了,现在我们开始学习进程了吧?
答:是的。前面说到线程就是我们的手,我们现在可以学习一下我们的“胳膊”了。
我们有了多线程,为什么还要学习多进程呢?这是因为在Python当中有一把GIL锁的存在,比如某些耗CPU的运算的时候,我们可以运行多进程多个CPU并发的操作进行操作。对于IO操作来说,我们的瓶颈不在于我们的CPU因此我们用多线程操作。进程切换操作不是轻量级的。
我们首先举例一个数据密集型的操作,来计算斐波那契数列:
from concurrent.futures import ThreadPoolExecutor,as_completed from concurrent.futures import ProcessPoolExecutor import time def fib(n): if n<=2: return 1 return fib(n-1) + fib(n-2) if __name__ == '__main__': with ThreadPoolExecutor(3) as executor: all_task = [executor.submit(fib,(num)) for num in range(25,40)] start_time = time.time() for future in as_completed(all_task): data = future.result() print("get result:= {}".format(data)) print("multithread last time is {}".format(time.time()-start_time)) with ProcessPoolExecutor(3) as executor: all_task = [executor.submit(fib,(num)) for num in range(25,40)] start_time = time.time() for future in as_completed(all_task): data = future.result() print("get result:= {}".format(data)) print("multiprocess last time is {}".format(time.time()-start_time)) # # multithread last time is 43.156678199768066 # multiprocess last time is 27.62783455848694
我们明显看到多进程比多线程快。
我们在以一个IO操作来进行对比:
from concurrent.futures import ThreadPoolExecutor,as_completed from concurrent.futures import ProcessPoolExecutor import time def random_sleep(n): time.sleep(n) return n if __name__ == '__main__': with ThreadPoolExecutor(3) as executor: all_task = [executor.submit(random_sleep,(num)) for num in [2]*30] start_time = time.time() for future in as_completed(all_task): data = future.result() print("get result:= {}".format(data)) print("multithread last time is {}".format(time.time()-start_time)) with ProcessPoolExecutor(3) as executor: all_task = [executor.submit(random_sleep,(num)) for num in [2]*30] start_time = time.time() for future in as_completed(all_task): data = future.result() print("get result:= {}".format(data)) print("multiprocess last time is {}".format(time.time()-start_time)) # # multithread last time is 20.035860300064087 # multiprocess last time is 20.641016483306885
正式进入我们的进程操作:
import os import time # fork只能用于linux下面 pid = os.fork() print("bobby") if pid == 0: print("子进程{},父进程是{}".format(os.getpid(),os.getppid())) else: print("我是父进程:{}".format(pid)) time.sleep(2)
这段代码只能在Linux下运行。我们发现的问题是如果主进程结束了,子进程还是会运行的。
问:进程如何进行编程?
答:我们懂了线程的编程,进程的编程会变得非常的简单。多余的内容就不再讲解,我们讲解一些不同的包,其实这些包的应用也是跟进程差不多的。
multiprocessing
import multiprocessing import time def get_html(n): time.sleep(n) return n if __name__ == '__main__': progress = multiprocessing.Process(target=get_html,args=(2,)) progress.start() progress.join()
我们还可以直接获取进程的pid和ppid。
其他和我们多线程差不都就不详解了。
使用进程池:
进程池:Pool和ProcessPoolExecutor。后那个跟线程一样。我们单独说一下Pool这个进程池。
import multiprocessing import time from multiprocessing import Pool def get_html(n): time.sleep(n) return n if __name__ == '__main__': progress = multiprocessing.Process(target=get_html,args=(1,)) progress.start() progress.join() pool = Pool(multiprocessing.cpu_count()) print(multiprocessing.cpu_count()) result = pool.apply_async(get_html,args=(3,)) pool.close()
注意最后要关闭线程池。详细的关于线程池的代码可以参照这里:https://www.cnblogs.com/noah0532/p/10938771.html
特别要说明的是有两个方法:imap 和 imap_unordered(这个是谁先完成先打印谁)
for result in pool.imap(get_html,[1,5,3]):
进程间的通信:
进程间的通信和线程间的通信有一样的也有不一样的地方,比如锁就不能使用了。
举一个简单的例子:用队列进行通信
from multiprocessing import Process,Queue # from queue import Queue # 这个queue就不能用了 import time def producer(queue): queue.put("a") time.sleep(2) def consumer(queue): time.sleep(2) data = queue.get() print(data) if __name__ == '__main__': queue = Queue(10) my_producer = Process(target=producer,args=(queue,)) my_consumer = Process(target=consumer, args=(queue,)) my_producer.start() my_consumer.start() my_producer.join() my_consumer.join()
在多进程的编程中不能用之前的queue了,带用multiprocessing里面的queue,这一带你要注意
我们再举一个共享变量的例子:
from multiprocessing import Process import time def producer(a): a += 1 time.sleep(2) def consumer(a): time.sleep(2) print(a) if __name__ == '__main__': a = 1 my_producer = Process(target=producer,args=(a,)) my_consumer = Process(target=consumer, args=(a,)) my_producer.start() my_consumer.start() my_producer.join() my_consumer.join()
我们发现我们的全局变量不能用了,正如我们前面说的,我们再进程中每一块的变量是单独的,不能共享的。
另外multiprocessing中的queue也不能用在进程池当中。如果我们想在进程当中应用就带用Manager当中的Queue
from multiprocessing import Process,Queue,Manager,Pool import time def producer(queue): queue.put("a") time.sleep(2) def consumer(queue): time.sleep(2) data = queue.get() print(data) if __name__ == '__main__': queue = Manager().Queue(10) pool = Pool(2) pool.apply_async(producer,args=(queue,)) pool.apply_async(consumer, args=(queue,)) pool.close() pool.join()
另外,我们还可以通过我们的pipe管道来进行通讯,但是Pipe只能使用两个进程间的通信,如果是两个交换pipe的性能比queue高
from multiprocessing import Process,Queue,Manager,Pool,Pipe import time def producer(pipe): pipe.send("bobby") def consumer(pipe): print(pipe.recv()) if __name__ == '__main__': # pipe只能用于两个进程间的通讯 receive_pipe,send_pipe = Pipe() my_producer = Process(target=producer,args=(send_pipe,)) my_consumer = Process(target=consumer, args=(receive_pipe,)) my_producer.start() my_consumer.start() my_producer.join() my_consumer.join()
重点:进程间的共享内存操作:Manager().dict(),array()....常用的数据类型都有。
from multiprocessing import Process,Queue,Manager,Pool,Pipe def add_data(p_dict,key,value): p_dict[key] = value if __name__ == '__main__': progress_dict = Manager().dict() first_progess = Process(target=add_data,args=(progress_dict,"bobby1",22)) second_progess = Process(target=add_data, args=(progress_dict, "bobby1", 23)) first_progess.start() second_progess.start() first_progess.join() second_progess.join() print(progress_dict) # {'bobby1': 23}