python 多进程多线程程序

这个纯粹为了增加理解，将很多比较好的资料进行归纳总结。

1、理论汇总

并发和并行

image.png
多进程和多线程

同步和异步
同步：所谓同步，就是在发出一个功能调用时，在没有得到结果之前，该调用就不会返回。

异步：异步的概念和同步相对，当一个异步功能调用发出后，调用者不能立即
得到结果。
简单来说，同步就是必须一件一件事做，等前一件做完了才能做下一件事。
同步和异步的区别：请求发出后，是否需要等待结果，才能继续执行其他操作。
假设说有个函数func(param),里面会有若干个模块，最后返回一个结果，同步就是说等这个函数调用结束返回值之后才会往下继续。意思就是说，这个调用里，这几个模块会紧挨着顺序进行。
而在异步中，这几个模块可能不是紧挨着的。
举个例子如下

什么是进程挂起
参加下面阻塞时候的情况。

什么是阻塞和非阻塞

阻塞和非阻塞这两个概念与程序（线程）等待消息通知(无所谓同步或者异步)时的状态有关。也就是说阻塞与非阻塞主要是程序（线程）等待消息通知时的状态角度来说的。

阻塞和非阻塞关注的是程序在等待调用结果（消息，返回值）时的状态.

阻塞调用是指调用结果返回之前，当前线程会被挂起。调用线程只有在得到结果之后才会返回。

非阻塞调用指在不能立刻得到结果之前，该调用不会阻塞当前线程。

关于异步、同步、阻塞和非阻塞的区别
同步/异步关注的是消息通知的机制，而阻塞/非阻塞关注的是程序（线程）等待消息通知时的状态。

以小明下载文件打个比方，从这两个关注点来再次说明这两组概念，希望能够更好的促进大家的理解。

同步阻塞：小明一直盯着下载进度条，到 100% 的时候就完成。

同步体现在：等待下载完成通知；

阻塞体现在：等待下载完成通知过程中，不能做其他任务处理；

同步非阻塞：小明提交下载任务后就去干别的，每过一段时间就去瞄一眼进度条，看到 100% 就完成。

同步体现在：等待下载完成通知，但是要在；

非阻塞体现在：等待下载完成通知过程中，去干别的任务了，只是时不时会瞄一眼进度条；【小明必须要在两个任务间切换，关注下载进度】

异步阻塞：小明换了个有下载完成通知功能的软件，下载完成就“叮”一声。不过小明仍然一直等待“叮”的声音（看起来很傻，不是吗）。

异步体现在：下载完成“叮”一声通知；

阻塞体现在：等待下载完成“叮”一声通知过程中，不能做其他任务处理；

异步非阻塞：仍然是那个会“叮”一声的下载软件，小明提交下载任务后就去干别的，听到“叮”的一声就知道完成了。

异步体现在：下载完成“叮”一声通知；

非阻塞体现在：等待下载完成“叮”一声通知过程中，去干别的任务了，只需要接收“叮”声通知即可；【软件处理下载任务，小明处理其他任务，不需关注进度，只需接收软件“叮”声通知，即可】

也就是说，同步/异步是“下载完成消息”通知的方式（机制），而阻塞/非阻塞则是在等待“下载完成消息”通知过程中的状态（能不能干其他任务），在不同的场景下，同步/异步、阻塞/非阻塞的四种组合都有应用。

所以，综上所述，同步和异步仅仅是关注的消息如何通知的机制，而阻塞与非阻塞关注的是等待消息通知时的状态。也就是说，同步的情况下，是由处理消息者自己去等待消息是否被触发，而异步的情况下是由触发机制来通知处理消息者，所以在异步机制中，处理消息者和触发机制之间就需要一个连接的桥梁：

在小明的例子中，这个桥梁就是软件“叮”的声音。

同步/异步与阻塞/非阻塞#

1 同步阻塞形式

效率是最低的，

拿上面的例子来说，就是你专心等待下载完成，什么别的事都不做。

实际程序中：就是未对fd 设置O_NONBLOCK标志位的read/write 操作；

2 异步阻塞形式

异步操作是可以被阻塞住的，只不过它不是在处理消息时阻塞，而是在等待消息通知时被阻塞。

3 同步非阻塞形式

实际上是效率低下的，

想象一下你一边干别的事情一边还需要抬头看下载完成没有，如果把干别的事情和观察下载完成情况的位置看成是程序的两个操作的话，这个程序需要在这两种不同的行为之间来回的切换，效率可想而知是低下的。

4 异步非阻塞形式

效率更高，

因为等待下载完成是你(等待者)的事情，而通知你则是电脑(消息触发机制)的事情，程序没有在两种不同的操作中来回切换。

至此，关于进程线程、同步异步、阻塞非阻塞、并发并行已经讲的差不多了，有讲的不好的地方请大佬指出。同时也谢谢网上大佬的文章帮助我理解了这些概念。

进程间的通信
进程共享内存
主要参加下面2.2 的例子

2、常用用法总结

2.1 最常用程序

在python中最常用的是用multiprocessing这个模块来实现多进程。

#coding: utf-8
import multiprocessing
import time

def func(msg):
    print "msg:", msg
    time.sleep(3)
    print "end"

if __name__ == "__main__":
    pool = multiprocessing.Pool(processes = 3)
    for i in xrange(4):
        msg = "hello %d" %(i)
        pool.apply_async(func, (msg, ))   #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去

    print "Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~"
    pool.close()
    pool.join()   #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
    print "Sub-process(es) done."

result.png

pool.apply_async(func, (msg, )) 是指异步加载（看上面解释）
针对结果，是说，每个进程在调用func 这个函数的时候，不需要等待一个进程调用结束，返回结果之后，再有新的进程调用，而是说，每个都可以进行调用，不用等到正在调用这个函数的进程调用结束再调用（有点绕）。通俗来说就是，msg:hello 1 和end 在输出中并不是紧挨着的。
pool.close() 是说不再加入新的进程
记住就好
pool.join() s是说等运行的所有进程结束之后才会运行后面的主程序。
这个体现在所有的子进程结束之后，才会输出“Sub-process(es) done”

2.2. demon

2.3 主进程和子进程共享内存

import multiprocessing 
from multiprocessing import Manager
# 查看电脑的核数
print(str(multiprocessing.cpu_count()))

def func(i,l):
    temp = [i,i+1]
    l.append(temp)
    print('%s series'%i)
   
if __name__ =='__main__':
    df = pd.DataFrame()
    manager = Manager()
    pool = multiprocessing.Pool(processes =4)
    l= manager.list([])
    for x in range(100):
#        pool.apply_async(func,(x,))
        pool.apply_async(func,(x,l))
#        df = df.append(temp)
        
    pool.close()
    pool.join()
    print(l)
    print('done')

result.png

首先设置一个manager = Manager()
在主进程中先设置一个l 作为list, l= manager.list([])
然后将l 传入每个子进程中pool.apply_async(func,(x,l))
然后在每个子进程中分别对l append
最后在主进程中l 就是所有的apend 加总。

2.4 进程池阻塞和非阻塞的代码

2.4.1. 阻塞

#coding: utf-8
import multiprocessing
import time

def func(msg):
    print "msg:", msg
    time.sleep(3)
    print "end"

if __name__ == "__main__":
    pool = multiprocessing.Pool(processes = 3)
    for i in xrange(4):
        msg = "hello %d" %(i)
        pool.apply(func, (msg, ))   #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去

    print "Mark~ Mark~ Mark~~~~~~~~~~~~~~~~~~~~~~"
    pool.close()
    pool.join()   #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束
    print "Sub-process(es) done."

image.png

2.4.2 非阻塞

import multiprocessing
import time

def func(msg):
    print "msg:", msg
    time.sleep(3)
    print "end"
    return "done" + msg

if __name__ == "__main__":
    pool = multiprocessing.Pool(processes=4)
    result = []
    for i in xrange(3):
        msg = "hello %d" %(i)
        result.append(pool.apply_async(func, (msg, )))
    pool.close()
    pool.join()
    for res in result:
        print ":::", res.get()
    print "Sub-process(es) done."