Python 中的多进程、多线程编程（关键词：Python/进程/线程/多进程/多线程）

时间  2020-06-10 标签 python 中 多 进程 多线程 编程 关键词 python 进程 线程 多 进程 多线程

进程

进程是什么？

进程：执行中的程序。包括代码段、程序计数器、堆栈段（临时数据，如局部变量、函数参数和返回地址）、堆（进程运行期间动态分配的内存）、进程状态、CPU 调度信息、I/O状态信息（如打开文件列表）python

进程控制块（Process Control Block）

（详见《操做系统概念（第 7 版）》 - P73）git

线程

线程是什么？

线程是 CPU 使用的基本单元，由线程 ID、程序计数器、寄存器集合和栈组成。github

进程与线程的区别？

线程与属于同一进程的其余线程共享代码段、公共数据段、打开文件和信号等等。web

在这里插入图片描述
（《操做系统概念（第 7 版）》 - P111）编程

多进程与多线程的区别？

区别 1：和线程不一样，进程没有任何共享状态；若是某个进程修改数据，改动只限于那个进程内。
（Python 参考手册 - P336）
区别 2：Python 中的多进程能够实现真正的并行，Python 中的多线程不能实现真正的并行。服务器

（OS概念 - 第 7 版 - P112）
线程优势：多线程

  1. 资源共享：线程默认共享所属进程的内存、代码、数据等资源；
  2. 经济：相比于进程，建立、切换线程的时间、资源开销较小；
  3. 响应度高。

Python 中的多线程？

因为 GIL 的存在，这个锁保证同一时刻只有 1 个线程在运行。
（《Python 核心编程(第 2 版)》 - 18.3.1 全局解释器锁 ( GIL ) - P514）
即便是多核 CPU，使用多线程编程，各个线程也只能交替使用 CPU 。
所以Python 中的多线程并不能实现真正的并行。
（Python 3 教程 - 廖雪峰 - 进程和线程 - 多线程 - 多核 CPU 。）app

Python 多进程编程

后台进程

若是 1 个进程是后台进程，那么，当建立这个后台进程的进程终止，后台进程会自动终止。
（Python 参考手册 - P337）dom

以单个进程的形式建立、启动函数

代码清单 - 1异步

# p337_single_process_with_function.py
from multiprocessing import Process
from time import ctime, sleep

def clock(interval):
        while True:
                print("The time is %s" % ctime())
                sleep(interval)

if __name__ == "__main__":
        p = Process(target=clock, args=(2,))
        #p.daemon = False
        p.start()
        #print("hello!")
        #p.join()

代码清单 - 2

# p337_single_process_with_function_Daemon_False.py
from multiprocessing import Process
from time import ctime, sleep

def clock(interval):
        while True:
                print("The time is %s" % ctime())
                sleep(interval)

if __name__ == "__main__":
        p = Process(target=clock, args=(2,))
        p.daemon = False
        p.start()
        #print("hello!")
        #p.join()

以上两段代码均会永远执行。

补充一点背景知识：

  1. 主进程的 daemon 标志默认为 False，即主进程默认不是后台进程；
  2. 在主进程开启的子进程会继承主进程的 daemon 标志，因而子进程的 daemon 标志为 False，子进程默认不是后台进程；
  3. 若是主进程结束了，后台进程也会随之结束，非后台进程不会随主进程结束而结束。

上面的两段代码的区别仅仅在因而否显式地将进程 p 设置为非后台进程，第 1 段代码虽然未将子进程 p 设置为非后台进程，可是，实际上效果等于第 2 段代码中 p.daemon = False，将子进程 p 的 daemon 标志设置为 False。
实际上能够认为两段代码是同样的。

下面仅解释第 2 段代码：
p = Process(target=clock, args=(2,)) 建立 1 个子进程 p，
p.daemon = False 将子进程 p 的daemon 标识设置为 False，
p.start() 以子进程 p 启动函数，
代码继续往下执行，主进程结束；因为子进程不是后台进程，所以，子进程 p 不会随主进程结束而结束，子进程 p 会继续运行，因为子进程 p 中的函数 clock 是死循环，所以子进程 p 会永远运行。

以下代码将子进程 p 的daemon 标志设置为 True：

代码清单 - 3

from multiprocessing import Process
from time import ctime, sleep

def clock(interval):
        while True:
                print("The time is %s" % ctime())
                sleep(interval)

if __name__ == "__main__":
        p = Process(target=clock, args=(2,))
        p.daemon = True
        p.start()
        #print("hello!")
        #p.join()

解释：
p.daemon = True 将子进程 p 设置为后台进程，所以，若是主进程结束，子进程 p （后台进程）也会随之结束；
p.start() 子进程 p 启动，执行clock函数；
代码继续往下执行，主进程结束，子进程 p 也随之结束（即便 clock 函数内部是死循环）。

参考文献：
Python 参考手册 - P337；
https://github.com/henry199101/python_CanKao_ShouCe/blob/master/chapter_20/p337_single_process_with_function.py；
https://github.com/henry199101/python_CanKao_ShouCe/blob/master/chapter_20/p337_single_process_with_function_Daemon_False.py；
https://github.com/henry199101/python_CanKao_ShouCe/blob/master/chapter_20/p337_single_process_with_function_Daemon_True.py

将进程定义为继承自 Process 的类

代码清单 - 4

from multiprocessing import Process
from time import ctime, sleep

class ClockProcess(Process):
        def __init__(self, interval):
                Process.__init__(self)
                self.interval = interval
        def run(self):
                while True:
                        print("The time is %s" % ctime())
                        sleep(self.interval)

if __name__ == "__main__":
        p = ClockProcess(2)
        p.start()

代码清单 4 中，进程 p 是非后台进程，与代码清单一、2相同，再也不作详细解释。

参考文献：
Python 参考手册 - P337
https://github.com/henry199101/python_CanKao_ShouCe/blob/master/chapter_20/p337_single_process_with_class.py。

进程间通讯

进程间通讯方式：管道、命名管道、消息队列、套接字、信号量、信号、共享内存、内存映射。
（https://blog.csdn.net/qq_33528613/article/details/77187572）
（unix进程间通讯方式(IPC)）

multiprocessing 模块支持的进程间通讯的 2 种主要形式：队列和管道。
（Python 参考手册 - P337）

队列（使用哨兵关闭消费者进程）

from multiprocessing import Process, Queue

def consumer(q):
        while True:
                item = q.get()

                if item is None:
                        break

                print(item)


def producer(sequence, q):
        for item in sequence:
                q.put(item)


if __name__ == "__main__":
        q = Queue()

        consumer_process = Process(target=consumer, args=(q,))

        consumer_process.start()

        sequence = [1, 2, 3, 4]
        producer(sequence, q)

        q.put(None)

        consumer_process.join()

（https://github.com/henry199101/python_CanKao_ShouCe/blob/master/chapter_20/p340_producer_consumer_shao_bing.py）
（Python 参考手册 - P340）

JoinableQueue

暂时略
（https://github.com/henry199101/python_CanKao_ShouCe/blob/master/chapter_20/p339_producer_consumer_JoinableQueue.py）
（Python 参考手册 - P339）

管道

在这里插入图片描述
在这里插入图片描述
APUE - 3rd - P431

from multiprocessing import Process, Pipe


def consumer(pipe):
        output_side, input_side = pipe

        input_side.close()

        while True:
                try:
                        item = output_side.recv()
                except EOFError:
                        break
                print(item)

        output_side.close()


def producer(sequence, pipe):
        output_side, input_side = pipe
        output_side.close()

        for item in sequence:
                input_side.send(item)

        input_side.close()


if __name__ == "__main__":
        pipe = Pipe()
        consumer_process = Process(target=consumer, args=(pipe,))
        consumer_process.start()

        sequence = [1,2,3,4]
        producer(sequence, pipe)

        consumer_process.join()

解释：

  1. Pipe()：建立 1 条管道，返回元组(conn1, conn2)；默认状况下是双向的；
  2. 若是生产者或者消费者没有使用管道的某个端点，就应该将其关闭；
  3. 管道是由操做系统进行引用计数的，必须在全部进程中关闭管道后（个人理解：引用计数减为 0 ），才能生成 EOFError 异常；
    （Python 参考手册 - P341——P343）

双向通讯的管道（使用管道，编写与进程交互的程序）

from multiprocessing import Process, Pipe

def adder(pipe):
        server_side, client_side = pipe

        client_side.close()

        while True:
                try:
                        x, y = server_side.recv()
                except EOFError:
                        break

                result = x + y

                server_side.send(result)

        print("Done!")


if __name__ == '__main__':
        pipe = Pipe()

        server_side, client_side = pipe

        adder_process = Process(target=adder, args=(pipe,))
        adder_process.start()

        server_side.close()

        client_side.send((3, 4))
        print(client_side.recv())

        client_side.send(('Hello, ', 'World!'))
        print(client_side.recv())

        client_side.close()

        adder_process.join()

（Python 参考手册 - P342——P343）

解释：
不作详细解释，请看图示，一目了然。

在这里插入图片描述

进程池

# 程序来自廖雪峰Python
#https://www.liaoxuefeng.com/wiki/0014316089557264a6b348958f449949df42a6d3a2e542c000/001431927781401bb47ccf187b24c3b955157bb12c5882d000#0

from multiprocessing import Pool
from time import time, sleep
from os import getpid
from random import random


def long_time_task(name):
        print("Task %s (%s) starts..." % (name, getpid()))
        start = time()
        sleep(random() * 3)
        end = time()
        elapsed_time = end - start
        print("Task %s runs %.2f seconds!" % (name, elapsed_time))


if __name__ == "__main__":
        print("Main process (%s) starts..." % getpid())
        p = Pool(4)
        for i in range(5):
                p.apply_async(long_time_task, args=(i,))
        print("Waiting for all subprocesses done!")
        p.close()
        p.join()
        print("All subprocesses done!")

输出：

$ python3 p343_p345_process_pool.py
Main process (2701) starts...
Waiting for all subprocesses done!
Task 0 (2702) starts...
Task 1 (2703) starts...
Task 3 (2704) starts...
Task 2 (2705) starts...
Task 2 runs 0.23 seconds!
Task 4 (2705) starts...
Task 3 runs 0.32 seconds!
Task 0 runs 0.64 seconds!
Task 1 runs 1.49 seconds!
Task 4 runs 1.75 seconds!
All subprocesses done!

解释：
p.apply_async() 在 1 个池进程中，异步地执行函数；
p.close() 用于关闭进程池；
p.join() 等待全部工做进程退出。
进程池开启了 4 个子进程，但却有 5 个子任务，所以最后 1 个子任务须要等到前面的 4 个进程中执行完 1 个，才能开始执行。

参考文献：
廖雪峰 - Python 3 - 多进程 - Pool
Python 参考手册 - P343——P345

链接

（Python 参考手册 - P352——P353）

服务器：

from multiprocessing.connection import Listener

server = Listener(address=('', 15000), authkey='12345')

while True:
        conn = server.accept()

        while True:
                try:
                        x, y = conn.recv()
                except EOFError:
                        break

                result = x + y

                conn.send(result)

        conn.close()

客户端：

from multiprocessing.connection import Client

conn = Client(address=('localhost', 15000), authkey='12345')

conn.send((2, 3))
result = conn.recv()
print(result)

conn.send(('Hello, ', 'World!'))
result = conn.recv()
print(result)

conn.close()

Python 多线程编程

以线程的形式建立、启动 1 个函数

from threading import Thread
from time import ctime, sleep

def clock(interval):
        while True:
                print("The time is %s" % ctime())
                sleep(interval)

t = Thread(target=clock, args=(2,))
t.daemon = True
t.start()

将单个线程定义为 1 个类

from threading import Thread
from time import sleep, ctime


class ClockThread(Thread):
        def __init__(self, seconds):
                Thread.__init__(self)
                self.seconds = seconds
                self.daemon = False
        def run(self):
                while True:
                        print('The time is %s' % ctime())
                        sleep(self.seconds)


if __name__ == '__main__':
        t = ClockThread(2)
        t.start()

Lock 对象

（Python 核心编程）

with 语句

信号量

可能暂时略

消息队列

concurrent.futures

联系我们 沪ICP备2021010478号-7