Python 的并发编程

happen 发布于2019-07-30 16:29 / 2645人阅读

摘要：本文最先发布在博客这篇文章将讲解并发编程的基本操作。并发是指能够多任务处理，并行则是是能够同时多任务处理。虽然自带了很好的类库支持多线程进程编程，但众所周知，因为的存在，很难做好真正的并行。

本文最先发布在博客：https://blog.ihypo.net/151628...

这篇文章将讲解 Python 并发编程的基本操作。并发和并行是对孪生兄弟，概念经常混淆。并发是指能够多任务处理，并行则是是能够同时多任务处理。Erlang 之父 Joe Armstrong 有一张非常有趣的图说明这两个概念：

我个人更喜欢的一种说法是：并发是宏观并行而微观串行。

GIL

虽然 Python 自带了很好的类库支持多线程/进程编程，但众所周知，因为 GIL 的存在，Python 很难做好真正的并行。

GIL 指全局解释器锁，对于 GIL 的介绍：

全局解释器锁（英语：Global Interpreter Lock，缩写GIL），是计算机程序设计语言解释器用于同步线程的一种机制，它使得任何时刻仅有一个线程在执行。

维基百科

其实与其说 GIL 是 Python 解释器的限制，不如说是 CPython 的限制，因为 Python 为了保障性能，底层大多使用 C 实现的，而 CPython 的内存管理并不是线程安全的，为了保障整体的线程安全，解释器便禁止多线程的并行执行。

因为 Python 社区认为操作系统的线程调度已经非常成熟了，没有必要自己再实现一遍，因此 Python 的线程切换基本是依赖操作系统，在实际的使用中，对于单核 CPU，GIL 并没有太大的影响，但对于多核 CPU 却引入了线程颠簸（thrashing）问题。

线程颠簸是指作为单一资源的 GIL 锁，在被多核心竞争强占时资源额外消耗的现象。

比如下图，线程1 在释放 GIL 锁后，操作系统唤醒了线程2，并将线程2 分配给核心2 执行，但是如果此时线程2 却没有成功获得 GIL 锁，只能再次被挂起。此时切换线程、切换上下文的资源都将白白浪费。

因此，Python 多线程程序在多核 CPU 机器下的性能不一定比单核高。那么如果是计算密集型的程序，一般还是考虑用 C 重写关键部分，或者使用多进程避开 GIL。

多线程

在 Python 中使用多线程，有 thread 和 threading 可供原则，thread 提供了低级别的、原始的线程以及一个简单的锁，因为 thread 过于简陋，线程管理容易出现人为失误，因此官方更建议使用 threading，而 threading 也不过是对 thread 的封装和补充。（Python3 中 thread 被改名为 _thread）。

在 Python 中创建线程非常简单：

import time
import threading


def do_task(task_name):
    print("Get task: {}".format(task_name))
    time.sleep(1)
    print("Finish task: {}".format(task_name))


if __name__ == "__main__":
    tasks = []
    for i in range(0, 10):
        # 创建 task
        tasks.append(threading.Thread(
            target=do_task,
            args=("task_{}".format(i),)))
    for t in tasks:
        # 开始执行 task
        t.start()

    for t in tasks:
        # 等待 task 执行完毕
        # 完毕前会阻塞住主线程
        t.join()
    print("Finish.")

直接创建线程简单优雅，如果逻辑复杂，也可以通过继承 Thread 基类完成多线程：

import time
import threading


class MyTask(threading.Thread):
    def __init__(self, task_name):
        super(MyTask, self).__init__()
        self.task_name = task_name

    def run(self):
        print("Get task: {}".format(self.task_name))
        time.sleep(1)
        print("Finish task: {}".format(self.task_name))


if __name__ == "__main__":
    tasks = []
    for i in range(0, 10):
        # 创建 task
        tasks.append(MyTask("task_{}".format(i)))
    for t in tasks:
        # 开始执行 task
        t.start()

    for t in tasks:
        # 等待 task 执行完毕
        # 完毕前会阻塞住主线程
        t.join()
    print("Finish.")

多进程

在 Python 中，可以使用 multiprocessing 库来实现多进程编程，和多线程一样，有两种方法可以使用多进程编程。

直接创建进程：

import time
import random
import multiprocessing


def do_something(task_name):
    print("Get task: {}".format(task_name))
    time.sleep(random.randint(1, 5))
    print("Finish task: {}".format(task_name))


if __name__ == "__main__":
    tasks = []
    for i in range(0, 10):
        # 创建 task
        tasks.append(multiprocessing.Process(
            target=do_something,
            args=("task_{}".format(i),)))
    for t in tasks:
        # 开始执行 task
        t.start()

    for t in tasks:
        # 等待 task 执行完毕
        # 完毕前会阻塞住主线程
        t.join()
    print("Finish.")

继承进程父类：

import time
import random
import multiprocessing


class MyTask(multiprocessing.Process):
    def __init__(self, task_name):
        super(MyTask, self).__init__()
        self.task_name = task_name

    def run(self):
        print("Get task: {}".format(self.task_name))
        time.sleep(random.randint(1, 5))
        print("Finish task: {}".format(self.task_name))


if __name__ == "__main__":
    tasks = []
    for i in range(0, 10):
        # 创建 task
        tasks.append(MyTask("task_{}".format(i)))
    for t in tasks:
        # 开始执行 task
        t.start()

    for t in tasks:
        # 等待 task 执行完毕
        # 完毕前会阻塞住主线程
        t.join()
    print("Finish.")

multiprocessing 除了常用的多进程编程外，我认为它最大的意义在于提供了一套规范，在该库下有一个 dummy 模块，即 multiprocessing.dummy，里面对 threading 进行封装，提供了和 multiprocessing 相同 API 的线程实现，换句话说，class::multiprocessing.Process 提供的是进程任务类，而 class::multiprocessing.dummy.Process，也正是有 multiprocessing.dummy 的存在，可以快速的讲一个多进程程序改为多线程：

import time
import random
from multiprocessing.dummy import Process


class MyTask(Process):
    def __init__(self, task_name):
        super(MyTask, self).__init__()
        self.task_name = task_name

    def run(self):
        print("Get task: {}".format(self.task_name))
        time.sleep(random.randint(1, 5))
        print("Finish task: {}".format(self.task_name))


if __name__ == "__main__":
    tasks = []
    for i in range(0, 10):
        # 创建 task
        tasks.append(MyTask("task_{}".format(i)))
    for t in tasks:
        # 开始执行 task
        t.start()

    for t in tasks:
        # 等待 task 执行完毕
        # 完毕前会阻塞住主线程
        t.join()
    print("Finish.")

无论是多线程还是多进程编程，这也是我一般会选择 multiprocessing 的原因。

除了直接创建进程，还可以用进程池（或者 multiprocessing.dummy 里的进程池）：

import time
import random
from multiprocessing import Pool


def do_task(task_name):
    print("Get task: {}".format(task_name))
    time.sleep(random.randint(1, 5))
    print("Finish task: {}".format(task_name))


if __name__ == "__main__":
    pool = Pool(5)
    for i in range(0, 10):
        #     创建 task
        pool.apply_async(do_task, ("task_{}".format(i),))
    pool.close()
    pool.join()
    print("Finish.")

线程池：

import time
import random
from multiprocessing.dummy import Pool


def do_task(task_name):
    print("Get task: {}".format(task_name))
    time.sleep(random.randint(1, 5))
    print("Finish task: {}".format(task_name))


if __name__ == "__main__":
    pool = Pool(5)
    for i in range(0, 10):
        #     创建 task
        pool.apply_async(do_task, ("task_{}".format(i),))
    pool.close()
    pool.join()
    print("Finish.")

这里示例有个问题，pool 在 join 前需要 close 掉，否则就会抛出异常，不过 Python 之禅的作者 Tim Peters 给出解释：

As to Pool.close(), you should call that when - and only when - you"re never going to submit more work to the Pool instance. So Pool.close() is typically called when the parallelizable part of your main program is finished. Then the worker processes will terminate when all work already assigned has completed.

It"s also excellent practice to call Pool.join() to wait for the worker processes to terminate. Among other reasons, there"s often no good way to report exceptions in parallelized code (exceptions occur in a context only vaguely related to what your main program is doing), and Pool.join() provides a synchronization point that can report some exceptions that occurred in worker processes that you"d otherwise never see.

同步原语

在多进程编程中，因为进程间的资源隔离，不需要考虑内存的线程安全问题，而在多线程编程中便需要同步原语来保存线程安全，因为 Python 是一门简单的语言，很多操作都是封装的操作系统 API，因此支持的同步原语蛮全，但这里只写两种常见的同步原语：锁和信号量。

通过使用锁可以用来保护一段内存空间，而信号量可以被多个线程共享。

在 threading 中可以看到 Lock 锁和 RLock 重用锁两种锁，区别如名。这两种锁都只能被一个线程拥有，第一种锁只能被获得一次，而重用锁可以被多次获得，但也需要同样次数的释放才能真正的释放。

当多个线程对同一块内存空间同时进行修改的时候，经常遇到奇怪的问题：

import time
import random
from threading import Thread, Lock

count = 0


def do_task():
    global count
    time.sleep(random.randint(1, 10) * 0.1)
    tmp = count
    tmp += 1
    time.sleep(random.randint(1, 10) * 0.1)
    count = tmp
    print(count)


if __name__ == "__main__":
    tasks = []
    for i in range(0, 10):
        tasks.append(Thread(target=do_task))
    for t in tasks:
        t.start()
    for t in tasks:
        t.join()
    print("Finish. Count = {}".format(count))

如上就是典型的非线程安全导致 count 没有达到预期的效果。而通过锁便可以控制某一段代码，或者说某段内存空间的访问：

import time
import random
from threading import Thread, Lock

count = 0
lock = Lock()


def do_task():
    lock.acquire()
    global count
    time.sleep(random.randint(1, 10) * 0.1)
    tmp = count
    tmp += 1
    time.sleep(random.randint(1, 10) * 0.1)
    count = tmp
    print(count)
    lock.release()


if __name__ == "__main__":
    tasks = []
    for i in range(0, 10):
        tasks.append(Thread(target=do_task))
    for t in tasks:
        t.start()
    for t in tasks:
        t.join()
    print("Finish. Count = {}".format(count))

当然，上述例子非常暴力，直接强行把并发改为串行。

对于信号量常见于有限资源强占的场景，可以定义固定大小的信号量供多个线程获取或者释放，从而控制线程的任务执行，比如下面的例子，控制最多有 5 个任务在执行：

import time
import random
from threading import Thread, BoundedSemaphore

sep = BoundedSemaphore(5)


def do_task(task_name):
    sep.acquire()
    print("do Task: {}".format(task_name))
    time.sleep(random.randint(1, 10))
    sep.release()


if __name__ == "__main__":
    tasks = []
    for i in range(0, 10):
        tasks.append(Thread(target=do_task, args=("task_{}".format(i),)))
    for t in tasks:
        t.start()
    for t in tasks:
        t.join()
    print("Finish.")

Queue 和 Pipe

因为多进程的内存隔离，不会存在内存竞争的问题。但同时，多个进程间的数据共享成为了新的问题，而进程间通信常见：队列，管道，信号。

这里只讲解队列和管道。

队列常见于双进程模型，一般用作生产者-消费者模式，由生产者进程向队列中发布任务，并由消费者从队列首部拿出任务进行执行：

import time
from multiprocessing import Process, Queue


class Task1(Process):
    def __init__(self, queue):
        super(Task1, self).__init__()
        self.queue = queue

    def run(self):
        item = self.queue.get()
        print("get item: [{}]".format(item))


class Task2(Process):
    def __init__(self, queue):
        super(Task2, self).__init__()
        self.queue = queue

    def run(self):
        print("put item: [Hello]")
        time.sleep(1)
        self.queue.put("Hello")


if __name__ == "__main__":
    queue = Queue()
    t1 = Task1(queue)
    t2 = Task2(queue)
    t1.start()
    t2.start()
    t1.join()
    print("Finish.")

理论上每个进程都可以向队列里的读或者写，可以认为队列是半双工路线。但是往往只有特定的读进程（比如消费者）和写进程（比如生产者），尽管这些进程只是开发者自己定义的。

而 Pipe 更像一个全工路线：

import time
from multiprocessing import Process, Pipe


class Task1(Process):
    def __init__(self, pipe):
        super(Task1, self).__init__()
        self.pipe = pipe

    def run(self):
        item = self.pipe.recv()
        print("Task1: recv item: [{}]".format(item))
        print("Task1: send item: [Hi]")
        self.pipe.send("Hi")


class Task2(Process):
    def __init__(self, pipe):
        super(Task2, self).__init__()
        self.pipe = pipe

    def run(self):
        print("Task2: send item: [Hello]")
        time.sleep(1)
        self.pipe.send("Hello")
        time.sleep(1)
        item = self.pipe.recv()
        print("Task2: recv item: [{}]".format(item))


if __name__ == "__main__":
    pipe = Pipe()
    t1 = Task1(pipe[0])
    t2 = Task2(pipe[1])
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("Finish.")

库

除了上面介绍的 threading 和 multiprocessing 两个库外，还有一个好用的令人发指的库 concurrent.futures。和前面两个库不同，这个库是更高等级的抽象，隐藏了很多底层的东西，但也因此非常好用。用官方的例子：

with ThreadPoolExecutor(max_workers=1) as executor:
    future = executor.submit(pow, 323, 1235)
    print(future.result())

该库中自带了进程池和线程池，可以通过上下文管理器来管理，而且对于异步任务执行完后，结果的获得也非常简单。再拿一个官方的多进程计算的例子作为结束：

import concurrent.futures
import math

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]

def is_prime(n):
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print("%d is prime: %s" % (number, prime))

if __name__ == "__main__":
    main()

欢迎关注个人公众号：CS实验室

GPU云服务器云服务器 Python并发编程并发编程 java并发编程 linux 并发编程

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/41698.html

使用 Python 进行并发编程系列 - 收藏集 - 掘金

摘要：使用进行并发编程篇三掘金这是使用进行并发编程系列的最后一篇。所以我考虑启用一个本地使用进行并发编程篇二掘金我们今天继续深入学习。使用 Python 进行并发编程 - asyncio 篇 (三) - 掘金这是「使用Python进行并发编程」系列的最后一篇。我特意地把它安排在了16年最后一天。重新实验上篇的效率对比的实现在第一篇我们曾经对比并发执行的效率，但是请求的是httpb...

MorePainMoreGain 2019-07-25 11:48 评论0 收藏0
Python基础之使用期物处理并发

摘要：本文重点掌握异步编程的相关概念了解期物的概念意义和使用方法了解中的阻塞型函数释放的特点。一异步编程相关概念阻塞程序未得到所需计算资源时被挂起的状态。导语：本文章记录了本人在学习Python基础之控制流程篇的重点知识及个人心得，打算入门Python的朋友们可以来一起学习并交流。本文重点： 1、掌握异步编程的相关概念；2、了解期物future的概念、意义和使用方法；3、了解Python...

asoren 2019-07-31 11:06 评论0 收藏0
从小白程序员一路晋升为大厂高级技术专家我看过哪些书籍？（建议收藏）

摘要：大家好，我是冰河有句话叫做投资啥都不如投资自己的回报率高。马上就十一国庆假期了，给小伙伴们分享下，从小白程序员到大厂高级技术专家我看过哪些技术类书籍。大家好，我是...

sf_wangchong 2021-09-29 09:35 评论0 收藏0
python初学——网络编程之FTP服务器支持多并发版本

摘要：扩展支持多用户并发访问与线程池。项目请见初学网络编程之服务器。不允许超过磁盘配额。该文件是一个使用模块编写的线程池类。这一步就做到了线程池的作用。对MYFTP项目进行升级。扩展支持多用户并发访问与线程池。MYFTP项目请见python初学——网络编程之FTP服务器。扩展需求 1.在之前开发的FTP基础上，开发支持多并发的功能2.不能使用SocketServer模块，必须自己实现多线...

oysun 2019-07-30 16:27 评论0 收藏0
史上最详细Python学习路线-从入门到精通，只需90天

摘要：针对的初学者，从无到有的语言如何入门，主要包括了的简介，如何下载，如何安装，如何使用终端，等各种开发环境进行开发，中的语法和基本知识概念和逻辑，以及继续深入学习的方法。 ...

gghyoo 2021-11-24 09:39 评论0 收藏0