进程和线程（上）

Python进阶者

发布于 2019-10-23 17:04:39

6040

发布于 2019-10-23 17:04:39

最近会开始继续 Python 的进阶系列文章，这是该系列的第一篇文章，介绍进程和线程的知识，刚好上一篇文章就介绍了采用 concurrent.futures 模块实现多进程和多线程的操作，本文则介绍下进程和线程的概念，多进程和多线程各自的实现方法和优缺点，以及分别在哪些情况采用多进程，或者是多线程。

因为文章比较长，所以会分为两篇进行介绍

概念

并发编程就是实现让程序同时执行多个任务，而如何实现并发编程呢，这里就涉及到进程和线程这两个概念。

对于操作系统来说，一个任务（或者程序）就是一个进程(Process)，比如打开一个浏览器是开启一个浏览器进程，打开微信就启动了一个微信的进程，打开两个记事本，就启动两个记事本进程。

进程的特点有：

操作系统以进程为单位分配存储空间， 每个进程有自己的地址空间、数据栈以及其他用于跟踪进程执行的辅助数据；
进程可以通过 fork 或者 spawn 方式创建新的进程来执行其他任务
进程都有自己独立的内存空间，所以进程需要通过进程间通信机制（IPC，Inter-Process Communication）来实现数据共享，具体的方式包括管道、信号、套接字、共享内存区等

一个进程还可以同时做多件事情，比如在 Word 里面同时进行打字、拼音检查、打印等事情，也就是一个任务分为多个子任务同时进行，这些进程内的子任务被称为线程（Thread）。

因为每个进程至少需要完成一件事情，也就是一个进程至少有一个线程。当要实现并发编程，也就是同时执行多任务时，有以下三种解决方案：

多进程，每个进程只有一个线程，但多个进程一起执行多个任务；
多线程，只启动一个进程，但一个进程内开启多个线程；
多进程+多线程，即启动多个进程，每个进程又启动多个线程，但这种方法非常复杂，实际很少使用

注意：真正的并行执行多任务只有在多核 CPU 上才可以实现，单核 CPU 系统中，真正的并发是不可能的，因为在某个时刻能够获得CPU的只有唯一的一个线程，多个线程共享了CPU的执行时间。

Python 是同时支持多进程和多线程的，下面就分别介绍多进程和多线程。

多进程

在 Unix/Linux 系统中，提供了一个 fork() 系统调用，它是一个特殊的函数，普通函数调用是调用一次，返回一次，但 fork 函数调用一次，返回两次，因为调用该函数的是父进程，然后复制出一份子进程了，最后同时在父进程和子进程内返回，所以会返回两次。

子进程返回的永远是 0 ，而父进程会返回子进程的 ID，因为父进程可以复制多个子进程，所以需要记录每个子进程的 ID，而子进程可以通过调用 getpid() 获取父进程的 ID。

Python 中 os 模块封装了常见的系统调用，这就包括了 fork ，代码示例如下：

import os

print('Process (%s) start...' % os.getpid())
# Only works on Unix/Linux/Mac:
pid = os.fork()
if pid == 0:
    print('I am child process (%s) and my parent is %s.' % (os.getpid(), os.getppid()))
else:
    print('I (%s) just created a child process (%s).' % (os.getpid(), pid))

运行结果：

Process (876) start...
I (876) just created a child process (877).
I am child process (877) and my parent is 876.

由于 windows 系统中是不存在 fork ，所以上述函数无法调用，但 Python 是跨平台的，所以也还是有其他模块可以实现多进程的功能，比如 multiprocessing模块。

multiprocess

multiprocessing 模块中提供了 Process 类来代表一个进程对象，接下来用一个下载文件的例子来说明采用多进程和不用多进程的差别。

首先是不采用多进程的例子：

def download_task(filename):
    '''模拟下载文件'''
    print('开始下载%s...' % filename)
    time_to_download = randint(5, 10)
    sleep(time_to_download)
    print('%s下载完成! 耗费了%d秒' % (filename, time_to_download))


def download_without_multiprocess():
    '''不采用多进程'''
    start = time()
    download_task('Python.pdf')
    download_task('nazha.mkv')
    end = time()
    print('总共耗费了%.2f秒.' % (end - start))
if __name__ == '__main__':
    download_without_multiprocess()

运行结果如下，这里用 randint 函数来随机输出当前下载文件的耗时，从结果看，程序运行时间等于两个下载文件的任务时间总和。

开始下载Python.pdf...
Python.pdf下载完成! 耗费了9秒
开始下载nazha.mkv...
nazha.mkv下载完成! 耗费了9秒
总共耗费了18.00秒.

如果是采用多进程，例子如下所示：

def download_task(filename):
    '''模拟下载文件'''
    print('开始下载%s...' % filename)
    time_to_download = randint(5, 10)
    sleep(time_to_download)
    print('%s下载完成! 耗费了%d秒' % (filename, time_to_download))

def download_multiprocess():
    '''采用多进程'''
    start = time()
    p1 = Process(target=download_task, args=('Python.pdf',))
    p1.start()
    p2 = Process(target=download_task, args=('nazha.mkv',))
    p2.start()
    p1.join()
    p2.join()
    end = time()
    print('总共耗费了%.2f秒.' % (end - start))
if __name__ == '__main__':
    download_multiprocess()

这里多进程例子中，我们通过 Process 类创建了进程对象，通过 target 参数传入一个函数表示进程需要执行的任务，args 是一个元组，表示传递给函数的参数，然后采用 start 来启动进程，而 join 方法表示等待进程执行结束。

运行结果如下所示，耗时就不是两个任务执行时间总和，速度上也是大大的提升了。

开始下载Python.pdf...
开始下载nazha.mkv...
Python.pdf下载完成! 耗费了5秒
nazha.mkv下载完成! 耗费了9秒
总共耗费了9.36秒.

Pool

上述例子是开启了两个进程，但如果需要开启大量的子进程，上述代码的写法就不合适了，应该采用进程池的方式批量创建子进程，还是用下载文件的例子，但执行下部分的代码如下所示：

import os
from multiprocessing import Process, Pool
from random import randint
from time import time, sleep

def download_multiprocess_pool():
    '''采用多进程，并用 pool 管理进程池'''
    start = time()
    filenames = ['Python.pdf', 'nazha.mkv', 'something.mp4', 'lena.png', 'lol.avi']
    # 进程池
    p = Pool(5)
    for i in range(5):
        p.apply_async(download_task, args=(filenames[i], ))
    print('Waiting for all subprocesses done...')
    # 关闭进程池
    p.close()
    # 等待所有进程完成任务
    p.join()
    end = time()
    print('总共耗费了%.2f秒.' % (end - start))
if __name__ == '__main__':
    download_multiprocess_pool()

代码中 Pool 对象先创建了 5 个进程，然后 apply_async 方法就是并行启动进程执行任务了，调用 join() 方法之前必须先调用 close() ，close() 主要是关闭进程池，所以执行该方法后就不能再添加新的进程对象了。然后 join() 就是等待所有进程执行完任务。

运行结果如下所示：

Waiting for all subprocesses done...
开始下载Python.pdf...
开始下载nazha.mkv...
开始下载something.mp4...
开始下载lena.png...
开始下载lol.avi...
nazha.mkv下载完成! 耗费了5秒
lena.png下载完成! 耗费了6秒
something.mp4下载完成! 耗费了7秒
Python.pdf下载完成! 耗费了8秒
lol.avi下载完成! 耗费了9秒
总共耗费了9.80秒.

子进程

大多数情况，子进程是一个外部进程，而非自身。在创建子进程后，我们还需要控制子进程的输入和输出。

subprocess 模块可以让我们很好地开启子进程以及管理子进程的输入和输出。

下面是演示如何用 Python 演示命令 nslookup www.python.org，代码如下所示：

import subprocess

print('$ nslookup www.python.org')
r = subprocess.call(['nslookup', 'www.python.org'])
print('Exit code:', r)

运行结果：

$ nslookup www.python.org
Server:        192.168.19.4
Address:    192.168.19.4#53

Non-authoritative answer:
www.python.org    canonical name = python.map.fastly.net.
Name:    python.map.fastly.net
Address: 199.27.79.223

Exit code: 0

如果子进程需要输入，可以通过 communicate() 进行输入，代码如下所示：

import subprocess

print('$ nslookup')
p = subprocess.Popen(['nslookup'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, err = p.communicate(b'set q=mx\npython.org\nexit\n')
print(output.decode('utf-8'))
print('Exit code:', p.returncode)

这段代码就是执行命令 nslookup 时，输入：

set q=mx
python.org
exit

运行结果：

$ nslookup
Server:        192.168.19.4
Address:    192.168.19.4#53

Non-authoritative answer:
python.org    mail exchanger = 50 mail.python.org.

Authoritative answers can be found from:
mail.python.org    internet address = 82.94.164.166
mail.python.org    has AAAA address 2001:888:2000:d::a6


Exit code: 0

进程间通信

进程之间是需要通信的，multiprocess 模块中也提供了 Queue、Pipes 等多种方式来交换数据。

这里以 Queue 为例，在父进程创建两个子进程，一个往 Queue 写入数据，另一个从 Queue 读取数据。代码如下：

import os
from multiprocessing import Process, Queue
import random
from time import time, sleep

# 写数据进程执行的代码:
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)
        sleep(random.random())


# 读数据进程执行的代码:
def read(q):
    print('Process to read: %s' % os.getpid())
    while True:
        value = q.get(True)
        print('Get %s from queue.' % value)


def ipc_queue():
    '''
    采用 Queue 实现进程间通信
    :return:
    '''
    # 父进程创建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()


if __name__ == '__main__':
    ipc_queue()

运行结果如下所示：

Process to write: 24992
Put A to queue...
Process to read: 22836
Get A from queue.
Put B to queue...
Get B from queue.
Put C to queue...
Get C from queue.

参考

https://www.liaoxuefeng.com/wiki/1016959663602400/1017627212385376
https://github.com/jackfrued/Python-100-Days/blob/master/Day01-15/13.%E8%BF%9B%E7%A8%8B%E5%92%8C%E7%BA%BF%E7%A8%8B.md
https://www.runoob.com/python3/python3-multithreading.html

本文主要是介绍进程和线程的概念，然后就是介绍多进程及其实现方式，在下一篇文章会介绍多线程的实现，以及两种方式应该如何选择。

代码已经上传到：

https://github.com/ccc013/Python_Notes/blob/master/Tutorials/Process_and_Threading/multi_process.py

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2019-10-21，如有侵权请联系 cloudcommunity@tencent.com 删除

编程算法