python多进程调用模块内函数_Python多线程、多进程（三）之线程进程对比、多进程...

一、多线程与多进程的对比

在之前简单的提过&＃xff0c;CPython中的GIL使得同一时刻只能有一个线程运行&＃xff0c;即并发执行。并且即使是多核CPU&＃xff0c;GIL使得同一个进程中的多个线程也无法映射到多个CPU上运行&＃xff0c;这么做最初是为了安全着想&＃xff0c;慢慢的也成为了限制CPython性能的问题。

一个线程想要执行&＃xff0c;就必须得到GIL&＃xff0c;否则就不能拿到CPU资源。但是也不是说一个线程在拿到CPU资源后就一劳永逸&＃xff0c;在执行的过程中GIL可能会释放并被其他线程获取&＃xff0c;所以说其它的线程会与本线程竞争CPU资源&＃xff0c;线程是抢占式执行的。具体可在 understand GIL中看到&＃xff0c;传送门。

多线程在python2中&＃xff1a;当一个线程进行I/O的时候会释放锁&＃xff0c;另外当ticks计数达到100&＃xff08;ticks可以看作是Python自身的一个计数器&＃xff0c;也可对比着字节码指令理解&＃xff0c;专门做用于GIL&＃xff0c;每次释放后归零&＃xff0c;这个计数可以通过 sys.setcheckinterval 来调整)。锁释放之后&＃xff0c;就涉及到线程的调度&＃xff0c;线程的锁进行&＃xff0c;线程的切换。这是会消耗CPU资源&＃xff0c;因此会造成程序性能问题和等待时延。另外由于线程共享内存的问题&＃xff0c;没有进程安全性高。

但是对于多进程&＃xff0c;GIL就无法限制&＃xff0c;多个进程可以再多个CPU上运行&＃xff0c;充分利用多核优势。事情往往是相对的&＃xff0c;虽然可以充分利用多核优势&＃xff0c;但是进程之的创建和调度却比线程的代价更高。

所以选择多线程还是多进程&＃xff0c;主要还是看怎样权衡代价&＃xff0c;什么样的情况。

1、CPU密集代码

下面来利用斐波那契数列模拟CPU密集运算。def fib(n):    # 求斐波那契数列的第n个值

if n<&＃61;2:        return 1

return fib(n-1)&＃43;fib(n-2)

<1>、多进程

打印第25到35个斐波那契数&＃xff0c;并计算程序运行时间import timefrom concurrent.futures import ThreadPoolExecutor, as_completedfrom concurrent.futures import ProcessPoolExecutordef fib(n):

if n<&＃61;2:        return 1

return fib(n-1)&＃43;fib(n-2)if __name__ &＃61;&＃61; "__main__":    with ProcessPoolExecutor(3) as executor:  # 使用进程池控制  每次执行3个进程

all_task &＃61; [executor.submit(fib, (num)) for num in range(25,35)]

start_time &＃61; time.time()        for future in as_completed(all_task):

data &＃61; future.result()

print("exe result: {}".format(data))

print("last time is: {}".format(time.time()-start_time))# 输出exe result: 75025exe result: 121393exe result: 196418exe result: 317811exe result: 514229exe result: 832040exe result: 1346269exe result: 2178309exe result: 3524578exe result: 5702887last time is: 4.457437038421631

输出结果&＃xff0c;每次打印三个exe result&＃xff0c;总重打印十个结果&＃xff0c;多进程运行时间为4.45秒

<2>、多线程import timefrom concurrent.futures import ThreadPoolExecutor, as_completedfrom concurrent.futures import ProcessPoolExecutordef fib(n):

if n<&＃61;2:        return 1

return fib(n-1)&＃43;fib(n-2)if __name__ &＃61;&＃61; "__main__":    with ThreadPoolExecutor(3) as executor:  # 使用线程池控制  每次执行3个线程

all_task &＃61; [executor.submit(fib, (num)) for num in range(25,35)]

start_time &＃61; time.time()        for future in as_completed(all_task):

data &＃61; future.result()

print("exe result: {}".format(data))

print("last time is: {}".format(time.time()-start_time))# 输出exe result: 121393exe result: 75025exe result: 196418exe result: 317811exe result: 514229exe result: 832040exe result: 1346269exe result: 2178309exe result: 3524578exe result: 5702887last time is: 7.3467772006988525

最终程序运行时间为7.34秒

程序的执行之间与计算机的性能有关&＃xff0c;每天计算机的执行时间都会有差异。从上述结果中看显然多线程比多进程要耗费时间。这就是因为对于密集代码(密集运算&＃xff0c;循环语句等)&＃xff0c;tick计数很快达到100&＃xff0c;GIL来回的释放竞争&＃xff0c;线程之间频繁切换&＃xff0c;所以对于密集代码的执行中&＃xff0c;多线程性能不如对进程。

2、I/O密集代码

一个线程在I/O阻塞的时候&＃xff0c;会释放GIL&＃xff0c;挂起&＃xff0c;然后其他的线程会竞争CPU资源&＃xff0c;涉及到线程的切换&＃xff0c;但是这种代价与较高时延的I/O来说是不足为道的。

下面用sleep函数模拟密集I/Odef random_sleep(n):

time.sleep(n)    return n

<1>、 多进程def random_sleep(n):    time.sleep(n)    return nif __name__ &＃61;&＃61; "__main__":

with ProcessPoolExecutor(5) as executor:

all_task &＃61; [executor.submit(random_sleep, (num)) for num in [2]*30]

start_time &＃61; time.time()        for future in as_completed(all_task):

data &＃61; future.result()            print("exe result: {}".format(data))        print("last time is: {}".format(time.time()-start_time))

#  输出

exe result: 2exe result: 2......&＃xff08;30个&＃xff09;

exe result: 2exe result: 2last time is: 12.412866353988647

每次打印5个结果&＃xff0c;总共二十个打印结果&＃xff0c;多进程运行时间为12.41秒

<2>、多线程def random_sleep(n):    time.sleep(n)    return nif __name__ &＃61;&＃61; "__main__":

with ThreadPoolExecutor(5) as executor:

all_task &＃61; [executor.submit(random_sleep, (num)) for num in [2]*30]

start_time &＃61; time.time()        for future in as_completed(all_task):

data &＃61; future.result()            print("exe result: {}".format(data))        print("last time is: {}".format(time.time()-start_time))

#  输出

exe result: 2exe result: 2......&＃xff08;30个&＃xff09;

exe result: 2exe result: 2last time is: 12.004231214523315

I/O密集多线程情况下&＃xff0c;程序的性能较多进程有了略微的提高。IO密集型代码(文件处理、网络爬虫等)&＃xff0c;多线程能够有效提升效率(单线程下有IO操作会进行IO等待&＃xff0c;造成不必要的时间浪费&＃xff0c;而开启多线程能在线程A等待时&＃xff0c;自动切换到线程B&＃xff0c;可以不浪费CPU的资源&＃xff0c;从而能提升程序执行效率)。所以python的多线程对IO密集型代码比较友好。

3、线程进程对比CPU密集型代码(各种循环处理、计数等等)&＃xff0c;多线程性能不如多进程。

I/O密集型代码(文件处理、网络爬虫等)&＃xff0c;多进程不如多线程。

二、多进程

在python 进程、线程 (一)已经有简单的进程介绍。

不过与多线程编程相比&＃xff0c;最需要注意的是这里多进程由并发执行变成了真正意义上的并行执行。

1、fork()调用

Unix/Linux操作系统提供了一个fork()系统调用&＃xff0c;它非常特殊。普通的函数调用&＃xff0c;调用一次&＃xff0c;返回一次&＃xff0c;但是fork()调用一次&＃xff0c;返回两次&＃xff0c;因为操作系统自动把当前进程&＃xff08;称为父进程&＃xff09;复制了一份&＃xff08;称为子进程&＃xff09;&＃xff0c;然后&＃xff0c;分别在父进程和子进程内返回。子进程永远返回0&＃xff0c;而父进程返回子进程的ID。这样做的理由是&＃xff0c;一个父进程可以fork出很多子进程&＃xff0c;所以&＃xff0c;父进程要记下每个子进程的ID&＃xff0c;而子进程只需要调用getppid()就可以拿到父进程的ID。Python的os模块封装了常见的系统调用&＃xff0c;其中就包括fork&＃xff0c;可以在Python程序中轻松创建子进程&＃xff0c;但是还是要有Unix/Linux系统支持&＃xff0c;windows没有系统调用fork()&＃xff0c;可以在本地虚拟机或者云服务器尝试&＃xff0c;默认liunx发行版中是有python2.X的。

情况一&＃xff1a;import osprint("Lanyu")  # 只打印一次

pid &＃61; os.fork()if pid &＃61;&＃61; 0:  print(&＃39;子进程 {} &＃xff0c;父进程是&＃xff1a; {}.&＃39; .format(os.getpid(), os.getppid()))else:  print(&＃39;我是父进程&＃xff1a;{}.&＃39;.format(os.getpid())

# 输出

Lanyu

我是父进程&＃xff1a;2993子进程2994&＃xff0c;父进程2993

fork()调用复制了一个进程&＃xff0c;然后程序中就有两个进程&＃xff0c;父进程的pid不为0&＃xff0c;所以先打印子进程2994&＃xff0c;父进程2993。然后子进程pid&＃61;0&＃xff0c;打印我是父进程&＃xff1a;2993。这里的Lanyu打印一次

情况二&＃xff1a;import ospid &＃61; os.fork()print("Lanyu")  # 这里打印两次if pid &＃61;&＃61; 0:  print(&＃39;子进程 {} &＃xff0c;父进程是&＃xff1a; {}.&＃39; .format(os.getpid(), os.getppid()))else:  print(&＃39;我是父进程&＃xff1a;{}.&＃39;.format(os.getpid())

# 输出

Lanyu

我是父进程&＃xff1a;2993Lanyu

子进程2994&＃xff0c;父进程2993

这里的Lanyu打印两次是因为&＃xff0c;由于fork()函数调用之后&＃xff0c;程序立即成成一个子进程&＃xff0c;主进程打印一次&＃xff0c;子进程再打印一次。因此这里的Lanyu打印两次。

情况三&＃xff1a;

还记得操作系统专业课的时候&＃xff0c;老师讲的一道考研题int main{    fork();    fork();    fork():    printf(&＃39;process&＃39;)    return 0;

}

三次fork()&＃xff0c;问此程序最终打印几个次process&＃xff0c;关键在于fork()函数的用途&＃xff0c;每一次都会复制一次进程&＃xff0c;则最终&＃xff0c;一个父进程被复制成8个进程&＃xff0c;打印8次。

2、python多进程

虽然python中没有提供直接的进程调用函数&＃xff0c;但是标准库中的模块提供能更多更方便的选择。 ProcessPoolExecutor进程池&＃xff0c;与 multiprocessing标准的多进程模块。其实ProcessPoolExecutor也是对multiprocessing的封装调用&＃xff0c;并且与ThreadPoolExecutor线程池提供的接口类似。而multiprocessing则更加底层。

<1>、进程编程import timeimport multiprocessingdef get_html(n):

time.sleep(n)

print("sub_progress success")    return nif __name__ &＃61;&＃61; "__main__":

progress &＃61; multiprocessing.Process(target&＃61;get_html, args&＃61;(2,))

print(progress.pid)  # 打印结果为None&＃xff0c;因为这个时候进程还未开启

progress.start()  # 进程开启

print(progress.pid)

progress.join()

print("main progress end")# 输出None5056sub_progress success

main progress end

<2>、使用进程池import timeimport multiprocessingdef get_html(n):

time.sleep(n)

print("sub_progress success")    return nif __name__ &＃61;&＃61; "__main__":    #使用进程池

pool &＃61; multiprocessing.Pool(multiprocessing.cpu_count())  # 可以指明进程数&＃xff0c;默认等于CPU数

result &＃61; pool.apply_async(get_html, args&＃61;(3,))    #等待所有任务完成

pool.close()

pool.join()

print(result.get())# 输出sub_progress success3

<3>、imap 接口

实例一&＃xff1a;import timeimport multiprocessingdef get_html(n):

time.sleep(n)

print("sub_progress success")    return nif __name__ &＃61;&＃61; "__main__":    # imap

for result in pool.imap(get_html, [1,5,3]):

print("{} sleep success".format(result))# 输出sub_progress success1 sleep success

sub_progress success

sub_progress success5 sleep success3 sleep success

imap有点像python提供的内置函数map&＃xff0c;讲[1,5,3]这个列表中的值一个一个传递给get_html函数对象&＃xff0c;并按照传值的先后顺序&＃xff0c;一一执行输出进程结果。

实例二&＃xff1a;import multiprocessing

import timedef get_html(n):

time.sleep(n)

print("sub_progress success")    return nif __name__ &＃61;&＃61; "__main__":

pool &＃61; multiprocessing.Pool(multiprocessing.cpu_count())  # 可以进程数&＃xff0c;不过最好是等于CPU数&＃xff0c;这里也是进程数

for result in pool.imap_unordered(get_html, [1,5,3]):

print("{} sleep success".format(result))# 输出sub_progress success1 sleep success

sub_progress success3 sleep success

sub_progress success5 sleep success

与imap方法不同的是imap_unordered方法&＃xff0c;imap_unordered是按照进程的执行完成的先后顺序&＃xff0c;打印进程执行结果&＃xff0c;而不是依照列表中的先后顺序。可以依照需要调用。

划重点**多进程编程中&＃xff0c;需要在__name__ &＃61;&＃61; __main__下编写**

更多API参考&＃xff1a;传送门

3、进程通信

<1>、共享变量通信

类比线程之间的通信&＃xff0c;首先想到的就是共享变量通信。但是在多进程中&＃xff0c;一个进程都有自的隔离区&＃xff0c;导致变量不能共享。

情况一&＃xff1a;def producer(a):

a &＃43;&＃61; 100

time.sleep(2)def consumer(a):

time.sleep(2)

print(a)if __name__ &＃61;&＃61; "__main__":

a &＃61; 1

my_producer &＃61; Process(target&＃61;producer, args&＃61;(a,))

my_consumer &＃61; Process(target&＃61;consumer, args&＃61;(a,))

my_producer.start()

my_consumer.start()

my_producer.join()

my_consumer.join()# 输出1

结果进程没有共享变量。

但是Python的标准模块提供了Manager()在内存中划出一块单独的内存区&＃xff0c;供所有的进程使用&＃xff0c;共享变量。

情况二&＃xff1a;from multiprocessing import Process&＃xff0c; Managerdef add_data(p_dict, key, value):

p_dict[key] &＃61; valueif __name__ &＃61;&＃61; "__main__":

progress_dict &＃61; Manager().dict()

first_progress &＃61; Process(target&＃61;add_data, args&＃61;(progress_dict, "666", 666))  # 更新progress_dict

second_progress &＃61; Process(target&＃61;add_data, args&＃61;(progress_dict, "999", 999))  # 更新progress_dict

first_progress.start()

second_progress.start()

first_progress.join()

second_progress.join()

print(progress_dict)# 打印结果{&＃39;666&＃39;: 666, &＃39;999&＃39;: 999}  # 实现了变量的共享

在Manager中还可以有其它的数据结构&＃xff0c;例如列表数组等可共享使用。

因此&＃xff0c;在使用多进程编程的时候&＃xff0c;如果像情况二共享全局变量&＃xff0c;就仍旧需要加锁实现进程同步。

<2>、Queue队列通信

在multiprocessing模块中有Queue类安全的队列&＃xff0c;也可以实现通信&＃xff0c;不过在这种情况下无法联通线程池。import timefrom multiprocessing import Process, Queue, Pool, Managerdef producer(queue):

queue.put("a")

time.sleep(2)def consumer(queue):

time.sleep(2)

data &＃61; queue.get()

print(data)if __name__ &＃61;&＃61; "__main__":

queue &＃61; Queue(10)  # 使用普通的Queue

pool &＃61; Pool(2)

pool.apply_async(producer, args&＃61;(queue,))

pool.apply_async(consumer, args&＃61;(queue,))

pool.close()

pool.join()# 无输出

想要使用进程池又实现消息队列通信就需要用到Manager管理者import timefrom multiprocessing import Process, Queue, Pool, Managerdef producer(queue):

queue.put("a")

time.sleep(2)def consumer(queue):

time.sleep(2)

data &＃61; queue.get()

print(data)if __name__ &＃61;&＃61; "__main__":

queue &＃61; Manager().Queue(10)  # 在使用Manger的时候需要先将Manager实例化在调用Queue

pool &＃61; Pool(2)

pool.apply_async(producer, args&＃61;(queue,))

pool.apply_async(consumer, args&＃61;(queue,))

pool.close()

pool.join()# 输出正常打印字符a

<3>、pipe管道通信

pipe也用于进程通信&＃xff0c;从功能上说&＃xff0c;提供的接口应该是queue的子集。但是queue为了更好的控制&＃xff0c;所以内部加了很多的锁&＃xff0c;而pipe在两个进程通信的时候性能会比queue更好一些。def producer(pipe):

pipe.send("Lanyu")def consumer(pipe):

print(pipe.recv())if __name__ &＃61;&＃61; "__main__":

recevie_pipe, send_pipe &＃61; Pipe()    #pipe只能适用于两个进程

my_producer&＃61; Process(target&＃61;producer, args&＃61;(send_pipe, ))

my_consumer &＃61; Process(target&＃61;consumer, args&＃61;(recevie_pipe,))

my_producer.start()

my_consumer.start()

my_producer.join()

my_consumer.join()# 输出Lanyu

三、总结

最开始为了引出GIL&＃xff0c;简单输了python源码的执行流程&＃xff0c;也是先编译成字节码再执行。在CPython中&＃xff0c;为了数据完整性和状态同步才有GIL&＃xff0c;GIL同样使得多线程不能利用CPU多核优势&＃xff0c;所以性能低部分是因为GIL。

线程需要加上GIL才能获取CPU资源&＃xff0c;才能执行。线程通信的时候&＃xff0c;可以用消息队列Queue和全局变量&＃xff0c;但是对于全局变量这种通信方式&＃xff0c;在执行字节码一定数量之后&＃xff0c;会释放GIL&＃xff0c;线程抢占式执行同样导致变量的混乱&＃xff0c;所以我们加上了用户级别的互斥锁Lock&＃xff0c;或者迭代锁Rlock保证了线程的状态同步。condition帮我们实现了线程的复杂通信&＃xff0c;而semaphore信号量&＃xff0c;使得我们在多个线程的情况下&＃xff0c;控制并发线程的数量。线程池进一步的封装&＃xff0c;提供了对线程的状态&＃xff0c;异步控制等操作。

对于多进程&＃xff0c;可以利用多核CPU优势&＃xff0c;但是使用多线程和多进程还需要进一步根据密集I/O和密集运算型代码等具体情况。多进程标准模块中提供的接口与多线程类似&＃xff0c;可相互参照。

陆陆续续总结关于这篇博文也有一个多星期了&＃xff0c;但是还是感觉有说不清楚的地方逻辑不通&＃xff0c;希望读者能在评论区指出。期间参阅了很多的文档&＃xff0c;博客&＃xff0c;教程。印象最深刻的还是Understand GIL这篇关于GIL的解释&＃xff0c;虽然是英文文档&＃xff0c;但是作者总是能以最精炼的句子表达最清晰的观点。