进程process和线程thread应用和区别—

Subprocess

subprocess主要是在Python中执行外部的程序和命令。在Python中，我们通过标准库中的subprocess包来fork一个子进程，并运行一个外部的程序。

subprocess包中定义有数个创建子进程的函数，这些函数分别以不同的方式创建子进程，所以我们可以根据需要来从中选取一个使用。另外subprocess还提供了一些管理标准流(standard stream)和管道(pipe)的工具，从而在进程间使用文本通信。

使用subprocess包中的函数创建子进程的时候，要注意3点:

    1) 在创建子进程之后，父进程是否暂停，并等待子进程运行。2) 函数返回什么3) 当returncode不为0时，父进程如何处理。

subprocess包中定义数个创建子进程的函数：

1、subprocess.call()
父进程等待子进程完成返回退出信息(returncode，相当于exit code)

2、subprocess.check_call()

父进程等待子进程完成
返回0
检查退出信息，如果returncode不为0，则举出错误subprocess.CalledProcessError，该对象包含有returncode属性，可用try...except...来检查。

3、subprocess.check_output()

父进程等待子进程完成
返回子进程向标准输出的输出结果
检查退出信息，如果returncode不为0，则举出错误subprocess.CalledProcessError，该对象包含有returncode属性和output属性，output属性为标准输出的输出结果，可用try...except...来检查。

三种函数使用方法类似，以subprocess.call()来说明:

import subprocess
rc = subprocess.call(["ls","-l"])

将程序名(ls)和所带的参数(-l)一起放在一个表中传递给subprocess.call()

可以通过一个shell来解释一整个字符串:

import subprocess
out = subprocess.call("ls -l", shell=True)
out = subprocess.call("cd ..", shell=True)

使用了shell=True这个参数。这个时候，我们使用一整个字符串，而不是一个表来运行子进程。Python将先运行一个shell，再用这个shell来解释这整个字符串。

Popen()

实际上，上面的三个函数都是基于Popen()的封装(wrapper)。封装的目的在于让我们容易使用子进程。
当想要更个性化需求时，就要转向Popen类，该类生成的对象用来代表子进程。

与上面的封装不同的是，Popen对象创建后，主程序不会自动等待子进程完成。必须调用对象的wait()方法，父进程才会等待 (也就是阻塞block)：

import subprocess
child = subprocess.Popen(["ping","-c","5","www.google.com"])
child.wait()
print("parent process")

还可以在父进程中对子进程进行其它操作，比如我们上面例子中的child对象:

child.poll()           # 检查子进程状态
child.kill()           # 终止子进程
child.send_signal()    # 向子进程发送信号
child.terminate()      # 终止子进程

子进程的PID存储在child.pid

子进程的文本流控制

(沿用child子进程) 子进程的标准输入，标准输出和标准错误也可以通过如下属性表示:

child.stdinchild.stdoutchild.stderr

可以在Popen()建立子进程时改变标准输入、标准输出和标准错误，并可以利用subprocess.PIPE将多个子进程的输入和输出连接在一起，构成管道(pipe):

import subprocesschild1 = subprocess.Popen(["ls","-l"], stdout=subprocess.PIPE)
child2 = subprocess.Popen(["wc"], stdin=child1.stdout,stdout=subprocess.PIPE)out = child2.communicate()print(out)

subprocess.PIPE实际上为文本流提供一个缓存区。child1的stdout将文本输出到缓存区，随后child2的stdin从该PIPE中将文本读取走。child2的输出文本也被存放在PIPE中，直到communicate()方法从PIPE中读取出PIPE中的文本。

注意⚠️：communicate()是Popen对象的一个方法，该方法会阻塞父进程，直到子进程完成。

还可以利用communicate()方法来使用PIPE给子进程输入:

通过使用subprocess包，我们可以运行外部程序。这极大的拓展了Python的功能。
如果你已经了解了操作系统的某些应用，你可以从Python中直接调用该应用(而不是完全依赖Python)，并将应用的结果输出给Python，并让Python继续处理。shell的功能(比如利用文本流连接各个应用)，就可以在Python中实现。

这个包有两个很大的局限性：

1)我们总是让subprocess运行外部的程序，而不是运行一个Python脚本内部编写的函数。
2)进程间只通过管道进行文本交流。以上限制了我们将subprocess包应用到更广泛的多进程任务。(这样的比较实际是不公平的，因为subprocessing本身就是设计成为一个shell，而不是一个多进程管理包)

Multiprocessing

multiprocessing包是Python中的多进程管理包,可以利用multiprocessing.Process对象来创建一个进程。

该进程可以运行在Python程序内部编写的函数，有start(), run(), join()的方法和Lock/Event/Semaphore/Condition类(这些对象可以像多线程那样，通过参数传递给各个进程)，用以同步进程，其用法与threading包中的同名类一致。
所以，multiprocessing的很大一部份与threading使用同一套API，只不过换到了多进程的情境。

但在使用这些共享API的时候，我们要注意以下几点:

在UNIX平台上，当某个进程终结之后，该进程需要被其父进程调用wait，否则进程成为僵尸进程(Zombie)。所以，有必要对每个Process对象调用join()方法 (实际上等同于wait)。对于多线程来说，由于只有一个进程，所以不存在此必要性。
multiprocessing提供了threading包中没有的IPC(比如Pipe和Queue)，效率上更高。应优先考虑Pipe和Queue，避免使用Lock/Event/Semaphore/Condition等同步方式 (因为它们占据的不是用户进程的资源)。
多进程应该避免共享资源。在多线程中，我们可以比较容易地共享资源，比如使用全局变量或者传递参数。在多进程情况下，由于每个进程有自己独立的内存空间，以上方法并不合适。此时我们可以通过共享内存和Manager的方法来共享资源。但这样做提高了程序的复杂度，并因为同步的需要而降低了程序的效率。

Process.PID中保存有PID，如果进程还没有start()，则PID为None。

multiprocessing模块也是跨平台的多进程模块。

下面的例子演示了启动一个子进程并等待其结束：

from multiprocessing import Process
import os#子进程要执行的代码def run_proc(name):print('Run child process %s (%s)...' % (name, os.getpid()))
if __name__ == '__main__':print('Parent process %s.' % os.getpid())p = Process(target=run_proc, args=('test',))#传入执行函数和 函数的参数，创建出一个实例print('Child process will start.')p.start()                                   #用start方法启动p.join()                                    #join方法可以等待子进程结束后再继续往下运行，，常用于进程间的同步print('Child process end.')#运行结果：
Parent process 67630.
Child process will start.
Run child process test (67631)...
Child process end.

创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动，这样创建进程比fork()还要简单。

Pipe和Queue

管道PIPE和消息队列message queue，multiprocessing包中有Pipe类和Queue类来分别支持这两种IPC机制。Pipe和Queue可以用来传送常见的对象。

1) Pipe可以是单向(half-duplex)，也可以是双向(duplex)。我们通过mutiprocessing.Pipe(duplex=False)创建单向管道 (默认为双向)。一个进程从PIPE一端输入对象，然后被PIPE另一端的进程接收，单向管道只允许管道一端的进程输入，而双向管道则允许从两端输入。
下面的程序展示了Pipe的使用:

#Multiprocessing with Pipeimport multiprocessing as muldef proc1(pipe):pipe.send('Hello')print('proc1 rec:', pipe.recv())def proc2(pipe):print('proc2 rec:', pipe.recv())pipe.send('hello, too')#Build a pipe
pipe = mul.Pipe()#这里的Pipe是双向的。#Pipe对象建立的时候，返回一个含有两个元素的表，每个元素代表Pipe的一端(Connection对象)。
# 我们对Pipe的某一端调用send()方法来传送对象，在另一端使用recv()来接收。#Pass an end of the pipe to process 1
p1 = mul.Process(target=proc1, args=(pipe[0],))
#Pass the another end of the pipe to process2
p2 = mul.Process(target=proc2, args=(pipe[1],))
p1.start()
p2.start()
p1.join()
p2.join()

2) Queue与Pipe相类似，都是先进先出的结构。但Queue允许多个进程放入，多个进程从队列取出对象。Queue使用mutiprocessing.Queue(maxsize)创建，maxsize表示队列中可以存放对象的最大数量。

下面的程序展示了Queue的使用:

from multiprocessing import Process, Queue
import os, time, random# 写数据进程执行的代码：
def write(q):print('Process to write: %s' % os.getpid())for value in ['A', 'B', 'C']:print('Put %s to queue...' % value)q.put(value)#放入队列time.sleep(random.random()) #random.random()返回随机生成的实数，范围：0-1;# time.sleep()推迟调用线程的执行# 读数据进程执行的代码：
def read(q):print('Process to read: %s' % os.getpid())while True:value = q.get(True)#get()方法获取 返回指定键的值print('Get %s from queue' % value)if __name__ == '__main__':#父进程创建Queue，并传给各个子进程：q = Queue()pw = Process(target=write, args=(q,))#往Queue里写数据;传入执行函数和 函数的参数，创建出一个实例pr = Process(target=read, args=(q,))#从Queue中读数据#启动子进程pw，写入：pw.start()#Process to write: 70124# Put A to queue...#启动子进程pr，读取：pr.start()#Process to read: 70125# Get A from queue#等待pw结束：pw.join()#join方法可以等待子进程结束后再继续往下运行，，常用于进程间的同步#write()进行循环，然后value='B'，把'B'放入了队列中，即写入了Queue中，产生结果：#Put B to queue...#Get B from queue#pr进程里是死循环，无法等待其结束，只能强行终止：pr.terminate()#'A'，'B', 'C'循环结束后，强行终止了#输出结果：
Process to write: 96363
Put A to queue...
Process to read: 96364
Get A from queue
Put B to queue...
Get B from queue
Put C to queue...
Get C from queue

Pool

进程池 (Process Pool)可以创建多个进程。这些进程就像是多个随时待命的士兵，准备执行任务(程序)。

import multiprocessing as mul
def f(x):return x**2pool = mul.Pool(5)
rel = pool.map(f, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
print(rel)

上面创建了一个容许5个进程的进程池 (Process Pool) 。Pool运行的每个进程都执行f()函数。我们利用map()方法，将f()函数作用到表的每个元素上。

除了map()方法外，Pool还有下面的常用方法：

apply_async(func,args)  从进程池中取出一个进程执行func，args为func的参数。它将返回一个AsyncResult的对象，你可以对该对象调用get()方法以获得结果。close()  进程池不再创建新的进程join()   wait进程池中的全部进程。必须对Pool先调用close()方法才能join。

from multiprocessing import Pool
import os, time, randomdef long_time_task(name):print('Run task %s (%s)...' % (name, os.getpid()))start = time.time()         #time.time()获取当前时间戳浮点数time.sleep(random.random() * 3) #random.random()返回随机生成的实数，范围：0-1;# time.sleep()推迟调用线程的执行end = time.time()print('Task %s runs %0.2f seconds' % (name, (end - start)))if __name__ == '__main__':print('Parent process %s.' % os.getpid())p = Pool(4)                                 #task0，1，2，3是立刻执行的，而task4是等待前面某个task完成后才执行for i in range(4):p.apply_async(long_time_task, args = (i,)) #pool.apply_async是异步方式调用：连续提交任务#pool.apply是同步方式调用：下一个task需要等待上一个task完成后才能开始运行print('Waitting for all subprocesses done...')p.close()                               #调用join()之前必须close，调用close()之后不能添加新的processp.join()                                #在对pool调用join()会等待所有子进程执行完毕#join()一般用于进程间的同步，join()可以等待子进程结束后继续运行print('All subprocesses done.’)#运行结果：
Parent process 96333.
Waitting for all subprocesses done...
Run task 0 (96334)...
Run task 1 (96335)...
Run task 2 (96336)...
Run task 3 (96337)...
Task 0 runs 0.09 seconds
Task 2 runs 1.87 seconds
Task 1 runs 2.62 seconds
Task 3 runs 2.65 seconds
All subprocesses done.

Threading包

Python主要通过标准库中的threading包来实现多线程。
启动一个线程就是把一个函数传入并创建Thread实例，然后调用start()开始执行：

import time, threading#新线程执行的代码
def loop():print('thread %s is running...' % threading.current_thread().name)#threading.current_thread().name，它永远返回当前线程的实例名字。n = 0while n < 5:n  = n + 1print('thread %s >>> %s' % (threading.current_thread().name, n))time.sleep(0.5)print('thread %s ended' % threading.current_thread().name)print('thread %s is running...' % threading.current_thread().name)#打印当前线程
t = threading.Thread(target=loop, name='LoopThread')#创建子线程实例，子线程名字为LoopThread
t.start()#启动子线程
t.join()
print('thread %s ended' % threading.current_thread().name)#最后打印当前线程MainThread关闭
'''
thread MainThread is running...#当前线程名称：MainThread
thread LoopThread is running...#创建子线程
thread LoopThread >>> 1
thread LoopThread >>> 2
thread LoopThread >>> 3
thread LoopThread >>> 4
thread LoopThread >>> 5
thread LoopThread ended
thread MainThread ended
'''

Python使用threading.Thread对象来代表线程，用threading.Lock对象来代表一个互斥锁 (mutex)。

在函数中使用global来声明变量为全局变量，从而让多线程共享i和lock(在C语言中，我们通过将变量放在所有函数外面来让它成为全局变量)。

#模拟多线程售票程序
import threading
import time
import osdef doChore():time.sleep(0.5)#方便以后改进程序，添加其他操作def booth(tid):global iglobal lock#声明i，lock为全局变量，从而让多线程共享i和lock#如果不声明，因i、lock是不可变对象，将会被当成一个局部变量；可变则不需声明。while True:lock.acquire()#只有一个线程可以Lock(获取lock)，继续执行代码，其他线程等待直到获取锁if i != 0:i = i - 1#售票，i储存剩余票数print(tid,':now left:', i)#剩票数doChore()#依然在Lock内部，故可以进行其他使用共享资源，如：打印剩票#time.sleep(0.5)等待了0.5s，代表额外的操作可能花费的时间else:print('Thread_id', tid, 'no more tickets')os._exit(0)#直接退出售票过程#sys.exit(n) 退出程序引发SystemExit异常, 可以捕获异常执行些清理工作. n默认值为0, 表示正常退出.#os._exit(n), 直接退出, 不抛异常, 不执行相关清理工作. 常用在子进程的退出.#exit()/quit(), 跑出SystemExit异常. 一般在交互式shell中退出时使用.lock.release()#除去障碍,释放锁doChore()#Lock此时已被释放，无法再使用共享资源，可以做一些不使用共享资源的事情（喝水，找零）#启动主要功能：
i = 100#总票数
lock = threading.Lock()#创建一个锁，即互斥锁；为了防止其他线程同时获取票数；同步线程对i的修改#开始10个线程
for k in range(10):new_thread = threading.Thread(target=booth,args=(k,))#设置线程；threading.Thread代表线程#target:传入（函数）来运行#args: 传入要调用的参数new_thread.start()

threading.Thread对象：我们已经介绍了该对象的start()和run(), 此外：

join()方法，调用该方法的线程将等待直到改Thread对象完成，再恢复运行。这与进程间调用wait()函数相类似。

下面的对象用于处理多线程同步。对象一旦被建立，可以被多个线程共享，并根据情况阻塞某些进程:

threading.Lock对象: mutex, 有acquire()和release()方法.threading.Condition对象: condition variable，建立该对象时，会包含一个Lock对象 (因为condition variable总是和mutex一起使用)。可以对Condition对象调用acquire()和release()方法，以控制潜在的Lock对象。此外:
* wait()方法，相当于cond_wait()
* notify_all()，相当与cond_broadcast()
* nofify()，与notify_all()功能类似，但只唤醒一个等待的线程，而不是全部threading.Semaphore对象: semaphore，也就是计数锁。建对象的时候，可以传递一个整数作为计数上限 (sema = threading.Semaphore(5))。它与Lock类似，也有Lock的两个方法。threading.Event对象:相当于没有潜在的Lock保护的condition variable。对象有True和False两个状态。可以多个线程使用wait()等待，直到某个线程调用该对象的set()方法，将对象设置为True。线程可以调用对象的clear()方法来重置对象为False状态。

死循环：

import threading, multiprocessingdef loop():x = 0while True:x = x ^ 1for i in range(multiprocessing.cpu_acunt()):t = threading.Thread(target=loop)t.start()

对Python解释器由于设计时有GIL全局锁，导致了多线程无法利用多核。

ThreadLocal

ThreadLocal本身是一个全局变量，但是每个线程却可以利用它来保存属于自己的私有数据，这些私有数据对其他线程也是不可见的。
最常用的地方就是为每个线程绑定一个数据库连接，HTTP请求，用户身份信息等，这样一个线程的所有调用到的处理函数都可以非常方便地访问这些资源。

import threadinglocal_school = threading.local()def process_student():std = local_school.studentprint('Hello, %s (in %s)' % (std, threading.current_thread().name))def process_thread(name):# 先执行传入绑定student,创建实例，name='Alica'：local_school.student = name#调用process_student()，线程name为'Thread-A'process_student()t1 = threading.Thread(target=process_thread, args=('Alica',), name='Thread-A’)#创建实例，传入线程name
t2 = threading.Thread(target=process_thread, args=('Bob',), name ='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()
#输出：
Hello, Alica (in Thread-A)
Hello, Bob (in Thread-B)

进程VS线程

首先，要实现多任务，通常我们会设计Master-Worker模式，Master负责分配任务，Worker负责执行任务，因此，多任务环境下，通常是一个Master，多个Worker。

线程切换

操作系统在切换进程或者线程时也是一样的，它需要先保存当前执行的现场环境（CPU寄存器状态、内存页等），然后，把新任务的执行环境准备好（恢复上次的寄存器状态，切换内存页等），才能开始执行。多任务一旦多到一个限度，就会消耗掉系统所有的资源，结果效率急剧下降，所有任务都做不好。

计算密集型 VS IO密集型
我们可以把任务分为计算密集型和IO密集型。
计算密集型任务由于主要消耗CPU资源，最好用C语言编写。
涉及到网络、磁盘IO的任务都是IO密集型任务，最合适的语言就是开发效率最高（代码量最少）的语言，脚本语言是首选，C语言最差。

异步IO
如果充分利用操作系统提供的异步IO支持，就可以用单进程单线程模型来执行多任务，这种全新的模型称为事件驱动模型，Nginx就是支持异步IO的Web服务器Python语言，单线程的异步编程模型称为协程

分布式进程
在Thread和Process中，应当优选Process，因为Process更稳定，而且，Process可以分布到多台机器上，而Thread最多只能分布到同一台机器的多个CPU上。
Python的multiprocessing模块不但支持多进程，其中managers子模块还支持把多进程分布到多台机器上。一个服务进程可以作为调度者，将任务分布到其他多个进程中，依靠网络通信。由于managers模块封装很好，不必了解网络通信的细节，就可以很容易地编写分布式多进程程序。
例子：如果我们已经有一个通过Queue通信的多进程程序在同一台机器上运行，现在，由于处理任务的进程任务繁重，希望把发送任务的进程和处理任务的进程分布到两台机器上。怎么用分布式进程实现？
原有的Queue可以继续使用，但是，通过managers模块把Queue通过网络暴露出去，就可以让其他机器的进程访问Queue了。
我们先看服务进程，服务进程负责启动Queue，把Queue注册到网络上，然后往Queue里面写入任务：

# task_master.pyimport random, time, queue
from multiprocessing.managers import BaseManager# 发送任务的队列:
task_queue = queue.Queue()
# 接收结果的队列:
result_queue = queue.Queue()# 从BaseManager继承的QueueManager:
class QueueManager(BaseManager):pass# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)
# 绑定端口5000, 设置验证码'abc':
manager = QueueManager(address=('', 5000), authkey=b'abc')
# 启动Queue:
manager.start()
# 获得通过网络访问的Queue对象:
task = manager.get_task_queue()
result = manager.get_result_queue()
# 放几个任务进去:
for i in range(10):n = random.randint(0, 10000)print('Put task %d...' % n)task.put(n)
# 从result队列读取结果:
print('Try get results...')
for i in range(10):r = result.get(timeout=10)print('Result: %s' % r)
# 关闭:
manager.shutdown()
print('master exit.')

在分布式多进程环境下，添加任务到Queue不可以直接对原始的task_queue进行操作，那样就绕过了QueueManager的封装，必须通过manager.get_task_queue()获得的Queue接口添加。
然后，在另一台机器上启动任务进程（本机上启动也可以）：

# task_worker.pyimport time, sys, queue
from multiprocessing.managers import BaseManager# 创建类似的QueueManager:
class QueueManager(BaseManager):pass# 由于这个QueueManager只从网络上获取Queue，所以注册时只提供名字:
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')# 连接到服务器，也就是运行task_master.py的机器:
server_addr = '127.0.0.1'
print('Connect to server %s...' % server_addr)
# 端口和验证码注意保持与task_master.py设置的完全一致:
m = QueueManager(address=(server_addr, 5000), authkey=b'abc')
# 从网络连接:
m.connect()
# 获取Queue的对象:
task = m.get_task_queue()
result = m.get_result_queue()
# 从task队列取任务,并把结果写入result队列:
for i in range(10):try:n = task.get(timeout=1)print('run task %d * %d...' % (n, n))r = '%d * %d = %d' % (n, n, n*n)time.sleep(1)result.put(r)except Queue.Empty:print('task queue is empty.')
# 处理结束:
print('worker exit.')
先启动task_master.py服务进程：
$ python3 task_master.py
Put task 3411...
Put task 1605...
Put task 1398...
Put task 4729...
Put task 5300...
Put task 7471...
Put task 68...
Put task 4219...
Put task 339...
Put task 7866...
Try get results...

task_master.py进程发送完任务后，开始等待result队列的结果。现在启动task_worker.py进程：

$ python3 task_worker.py
Connect to server 127.0.0.1...
run task 3411 * 3411...
run task 1605 * 1605...
run task 1398 * 1398...
run task 4729 * 4729...
run task 5300 * 5300...
run task 7471 * 7471...
run task 68 * 68...
run task 4219 * 4219...
run task 339 * 339...
run task 7866 * 7866...
worker exit.

task_worker.py进程结束，在task_master.py进程中会继续打印出结果

Result: 3411 * 3411 = 11634921
Result: 1605 * 1605 = 2576025
Result: 1398 * 1398 = 1954404
Result: 4729 * 4729 = 22363441
Result: 5300 * 5300 = 28090000
Result: 7471 * 7471 = 55815841
Result: 68 * 68 = 4624
Result: 4219 * 4219 = 17799961
Result: 339 * 339 = 114921
Result: 7866 * 7866 = 61873956

其实这就是一个简单但真正的分布式计算，把代码稍加改造，启动多个worker，就可以把任务分布到几台甚至几十台机器上，比如把计算n*n的代码换成发送邮件，就实现了邮件队列的异步发送。