关于python:使用Python进行并发编程

35次阅读

共计 9021 个字符，预计需要花费 23 分钟才能阅读完成。

让计算机程序并发的运行是一个常常被探讨的话题，明天我想讨论一下 Python 下的各种并发形式。

并发形式

线程（[Thread]）

多线程简直是每一个程序猿在应用每一种语言时都会首先想到用于解决并发的工具（JS 程序员请回避），应用多线程能够无效的利用 CPU 资源（Python 例外）。然而多线程所带来的程序的复杂度也不可避免，尤其是对竞争资源的同步问题。

然而在 python 中因为应用了全局解释锁（GIL）的起因，代码并不能同时在多核上并发的运行，也就是说，Python 的多线程不能并发，很多人会发现应用多线程来改良本人的 Python 代码后，程序的运行效率却降落了，这是如许蛋疼的一件事呀！实际上应用多线程的编程模型是很艰难的，程序员很容易犯错，这并不是程序员的谬误，因为并行思维是反人类的，咱们大多数人的思维是串行（精神分裂不探讨），而且冯诺依曼设计的计算机架构也是以程序执行为根底的。所以如果你总是不能把你的多线程程序搞定，祝贺你，你是个思维失常的程序猿：）

Python 提供两组线程的接口，一组是 thread 模块，提供根底的，低等级（Low Level）接口，应用 Function 作为线程的运行体。还有一组是 threading 模块，提供更容易应用的基于对象的接口（相似于 Java），能够继承 Thread 对象来实现线程，还提供了其它一些线程相干的对象，例如 Timer，Lock

应用 thread 模块的例子 

import thread
def worker():
    """thread worker function"""
    print 'Worker'
thread.start_new_thread(worker)

应用 threading 模块的例子 

import threading
def worker():
    """thread worker function"""
    print 'Worker'
t = threading.Thread(target=worker)
t.start()

或者 Java Style

import threading
class worker(threading.Thread):
    def __init__(self):
        pass
    def run():
        """thread worker function"""
        print 'Worker'
    
t = worker()
t.start()

过程（Process）

因为前文提到的全局解释锁的问题，Python 下比拟好的并行形式是应用多过程，这样能够十分无效的应用 CPU 资源，并实现真正意义上的并发。当然，过程的开销比线程要大，也就是说如果你要创立数量惊人的并发过程的话，须要考虑一下你的机器是不是有一颗弱小的心。

Python 的 mutliprocess 模块和 threading 具备相似的接口。

from multiprocessing import Process

def worker():
    """thread worker function"""
    print 'Worker'
p = Process(target=worker)
p.start()
p.join()

因为线程共享雷同的地址空间和内存，所以线程之间的通信是非常容易的，然而过程之间的通信就要简单一些了。常见的过程间通信有，管道，音讯队列，Socket 接口（TCP/IP）等等。

Python 的 mutliprocess 模块提供了封装好的管道和队列，能够不便的在过程间传递音讯。

Python 过程间的同步应用锁，这一点喝线程是一样的。

另外，Python 还提供了过程池 Pool 对象，能够不便的治理和控制线程。

近程分布式主机（Distributed Node）

随着大数据时代的到临，摩尔定理在单机上仿佛曾经失去了成果，数据的计算和解决须要分布式的计算机网络来运行，程序并行的运行在多个主机节点上，曾经是当初的软件架构所必须思考的问题。

近程主机间的过程间通信有几种常见的形式

  • TCP/IP

    TCP/IP 是所有近程通信的根底，然而 API 比拟低级别，应用起来比拟繁琐，所以个别不会思考

  • 近程办法调用 Remote Function Call

    [RPC]

  • 近程对象 Remote Object

    近程对象是更高级别的封装，程序能够想操作本地对象一样去操作一个近程对象在本地的代理。近程对象最广为应用的标准 CORBA，CORBA 最大的益处是能够在不同语言和平台中进行通信。当让不必的语言和平台还有一些各自的近程对象实现，例如 Java 的 RMI，MS 的 DCOM

    Python 的开源实现，有许多对近程对象的反对

  • Dopy]
  • Fnorb（CORBA）
  • ICE
  • omniORB（CORBA）
  • Pyro
  • YAMI

  • 音讯队列 Message Queue

    比起 RPC 或者近程对象，音讯是一种更为灵便的通信伎俩，常见的反对 Python 接口的音讯机制有

  • RabbitMQ
  • ZeroMQ
  • Kafka
  • AWS SQS + BOTO

在近程主机上执行并发和本地的多过程并没有十分大的差别，都须要解决过程间通信的问题。当然对近程过程的治理和协调比起本地要简单。

Python 下有许多开源的框架来反对分布式的并发，提供无效的管理手段包含：

  • Celery

    Celery 是一个十分成熟的 Python 分布式框架，能够在分布式的零碎中，异步的执行工作，并提供无效的治理和调度性能。

  • SCOOP

    SCOOP（Scalable COncurrent Operations in Python）提供简略易用的分布式调用接口，应用 Future 接口来进行并发。

  • Dispy

    相比起 Celery 和 SCOOP，Dispy 提供更为轻量级的分布式并行服务

  • PP

    PP（Parallel Python）是另外一个轻量级的 Python 并行服务

  • Asyncoro

    Asyncoro 是另一个利用 Generator 实现分布式并发的 Python 框架，

当然还有许多其它的零碎，我没有一一列出

另外，许多的分布式系统多提供了对 Python 接口的反对，例如 Spark

伪线程（Pseudo-Thread）

还有一种并发伎俩并不常见，咱们能够称之为伪线程，就是看上去像是线程，应用的接口相似线程接口，然而理论应用非线程的形式，对应的线程开销也不存的。

  • greenlet

    greenlet 提供轻量级的 coroutines 来反对过程内的并发。

    greenlet 是 Stackless 的一个副产品，应用 tasklet 来反对一中被称之为微线程（mirco-thread）的技术，这里是一个应用 greenlet 的伪线程的例子 

from greenlet import greenlet

def test1():
    print 12
    gr2.switch()
    print 34
    
def test2():
    print 56
    gr1.switch()
    print 78
    
gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()
    运行以上程序失去如下后果：
12
56
34

伪线程 gr1 switch 会打印 12，而后调用 gr2 switch 失去 56，而后 switch 回到 gr1，打印 34，而后伪线程 gr1 完结，程序退出，所以 78 永远不会被打印。通过这个例子咱们能够看出，应用伪线程，咱们能够无效的控制程序的执行流程，然而伪线程并不存在真正意义上的并发。

eventlet，gevent 和 concurence 都是基于 greenlet 提供并发的。

  • eventlet http://eventlet.net/

eventlet 是一个提供网络调用并发的 Python 库，使用者能够以非阻塞的形式来调用阻塞的 IO 操作。

import eventlet
from eventlet.green import urllib2

urls = ['http://www.google.com', 'http://www.example.com', 'http://www.python.org']

def fetch(url):
    return urllib2.urlopen(url).read()

pool = eventlet.GreenPool()

for body in pool.imap(fetch, urls):
    print("got body", len(body))

执行后果如下 

('got body', 17629)
('got body', 1270)
('got body', 46949)

eventlet 为了反对 generator 的操作对 urllib2 做了批改，接口和 urllib2 是统一的。这里的 GreenPool 和 Python 的 Pool 接口统一。

  • gevent

gevent 和 eventlet 相似，

import gevent
from gevent import socket
urls = ['www.google.com', 'www.example.com', 'www.python.org']
jobs = [gevent.spawn(socket.gethostbyname, url) for url in urls]
gevent.joinall(jobs, timeout=2)

print [job.value for job in jobs]

执行后果如下：

['206.169.145.226', '93.184.216.34', '23.235.39.223']
  • concurence
    https://github.com/concurrenc…

concurence 是另外一个利用 greenlet 提供网络并发的开源库，我没有用过，大家能够本人尝试一下。

实战使用

通常须要用到并发的场合有两种，一种是计算密集型，也就是说你的程序须要大量的 CPU 资源; 另一种是 IO 密集型，程序可能有大量的读写操作，包含读写文件，收发网络申请等等。

计算密集型

对应计算密集型的利用，咱们选用驰名的蒙特卡洛算法来计算 PI 值。基本原理如下

蒙特卡洛算法利用统计学原理来模拟计算圆周率，在一个正方形中，一个随机的点落在 1 / 4 圆的区域（红色点）的概率与其面积成正比。也就该概率 p = Pi ＊ R＊R /4：R＊ R，其中 R 是正方形的边长，圆的半径。也就是说该概率是圆周率的 1 /4, 利用这个论断，只有咱们模仿出点落在四分之一圆上的概率就能够晓得圆周率了，为了失去这个概率，咱们能够通过大量的试验，也就是生成大量的点，看看这个点在哪个区域，而后统计出后果。

根本算法如下：

from math import hypot
from random import random

def test(tries):
    return sum(hypot(random(), random()) < 1 for _ in range(tries))

这里 test 办法做了 n（tries）次试验，返回落在四分之一圆中的点的个数。判断办法是查看该点到圆心的间隔，如果小于 R 则是在圆上。

通过大量的并发，咱们能够疾速的运行屡次试验，试验的次数越多，后果越靠近实在的圆周率。

这里给出不同并发办法的程序代码

  • 非并发

    咱们先在单线程，但过程运行，看看性能如何 

from math import hypot
from random import random
import eventlet
import time

def test(tries):
    return sum(hypot(random(), random()) < 1 for _ in range(tries))

def calcPi(nbFutures, tries):
    ts = time.time()
    result = map(test, [tries] * nbFutures)
    
    ret = 4. * sum(result) / float(nbFutures * tries)
    span = time.time() - ts
    print "time spend", span
    return ret

print calcPi(3000,4000)

  • 多线程 thread

    为了应用线程池，咱们用 multiprocessing 的 dummy 包，它是对多线程的一个封装。留神这里代码尽管一个字的没有提到线程，但它千真万确是多线程。

    通过测试咱们开（jing）心（ya）的发现，果然不出所料，当线程池为 1 是，它的运行后果和没有并发时一样，当咱们把线程池数字设置为 5 时，耗时简直是没有并发的 2 倍，我的测试数据从 5 秒到 9 秒。所以对于计算密集型的工作，还是放弃多线程吧。

from multiprocessing.dummy import Pool

from math import hypot
from random import random
import time

def test(tries):
    return sum(hypot(random(), random()) < 1 for _ in range(tries))

def calcPi(nbFutures, tries):
    ts = time.time()
    p = Pool(1)
    result = p.map(test, [tries] * nbFutures)
    ret = 4. * sum(result) / float(nbFutures * tries)
    span = time.time() - ts
    print "time spend", span
    return ret

if __name__ == '__main__':
    p = Pool()
    print("pi = {}".format(calcPi(3000, 4000)))

  • 多过程 multiprocess

    实践上对于计算密集型的工作，应用多过程并发比拟适合，在以下的例子中，过程池的规模设置为 5，批改过程池的大小能够看到对后果的影响，当过程池设置为 1 时，和多线程的后果所需的工夫相似，因为这时候并不存在并发；当设置为 2 时，响应工夫有了显著的改良，是之前没有并发的一半；然而持续扩充过程池对性能影响并不大，甚至有所降落，兴许我的 Apple Air 的 CPU 只有两个核？

    当心，如果你设置一个十分大的过程池，你会遇到 Resource temporarily unavailable 的谬误，零碎并不能反对创立太多的过程，毕竟资源是无限的。

from multiprocessing import Pool

from math import hypot
from random import random
import time

def test(tries):
    return sum(hypot(random(), random()) < 1 for _ in range(tries))

def calcPi(nbFutures, tries):
    ts = time.time()
    p = Pool(5)
    result = p.map(test, [tries] * nbFutures)
    ret = 4. * sum(result) / float(nbFutures * tries)
    span = time.time() - ts
    print "time spend", span
    return ret

if __name__ == '__main__':
    print("pi = {}".format(calcPi(3000, 4000)))

  • gevent（伪线程）

    不论是 gevent 还是 eventlet，因为不存在理论的并发，响应工夫和没有并发区别不大，这个和测试后果统一。

import gevent
from math import hypot
from random import random
import time

def test(tries):
    return sum(hypot(random(), random()) < 1 for _ in range(tries))

def calcPi(nbFutures, tries):
    ts = time.time()
    jobs = [gevent.spawn(test, t) for t in [tries] * nbFutures]
    gevent.joinall(jobs, timeout=2)
    ret = 4. * sum([job.value for job in jobs]) / float(nbFutures * tries)
    span = time.time() - ts
    print "time spend", span
    return ret

print calcPi(3000,4000)
  • eventlet（伪线程）
from math import hypot
from random import random
import eventlet
import time

def test(tries):
    return sum(hypot(random(), random()) < 1 for _ in range(tries))

def calcPi(nbFutures, tries):
    ts = time.time()
    pool = eventlet.GreenPool()
    result = pool.imap(test, [tries] * nbFutures)
    
    ret = 4. * sum(result) / float(nbFutures * tries)
    span = time.time() - ts
    print "time spend", span
    return ret

print calcPi(3000,4000)
  • SCOOP

SCOOP 中的 Future 接口合乎 PEP-3148 的定义，也就是在 Python3 中提供的 Future 接口。

在缺省的 SCOOP 配置环境下（单机，4 个 Worker），并发的性能有进步，然而不如两个过程池配置的多过程。

from math import hypot
from random import random
from scoop import futures

import time

def test(tries):
    return sum(hypot(random(), random()) < 1 for _ in range(tries))

def calcPi(nbFutures, tries):
    ts = time.time()
    expr = futures.map(test, [tries] * nbFutures)
    ret = 4. * sum(expr) / float(nbFutures * tries)
    span = time.time() - ts
    print "time spend", span
    return ret

if __name__ == "__main__":
    print("pi = {}".format(calcPi(3000, 4000)))
  • Celery

工作代码 

from celery import Celery

from math import hypot
from random import random
 
app = Celery('tasks', backend='amqp', broker='amqp://guest@localhost//')
app.conf.CELERY_RESULT_BACKEND = 'db+sqlite:///results.sqlite'
 
@app.task
def test(tries):
    return sum(hypot(random(), random()) < 1 for _ in range(tries))

客户端代码 

from celery import group
from tasks import test

import time

def calcPi(nbFutures, tries):
    ts = time.time()
    result = group(test.s(tries) for i in xrange(nbFutures))().get()
    
    ret = 4. * sum(result) / float(nbFutures * tries)
    span = time.time() - ts
    print "time spend", span
    return ret

print calcPi(3000, 4000)

应用 Celery 做并发的测试后果出其不意（环境是单机，4frefork 的并发，音讯 broker 是 rabbitMQ），是所有测试用例里最蹩脚的，响应工夫是没有并发的 5～6 倍。这兴许是因为管制协调的开销太大。对于这样的计算工作，Celery 兴许不是一个好的抉择。

  • asyncoro

    Asyncoro 的测试后果和非并发保持一致。

import asyncoro

from math import hypot
from random import random
import time

def test(tries):
    yield sum(hypot(random(), random()) < 1 for _ in range(tries))


def calcPi(nbFutures, tries):
    ts = time.time()
    coros = [asyncoro.Coro(test,t) for t in [tries] * nbFutures]
    ret = 4. * sum([job.value() for job in coros]) / float(nbFutures * tries)
    span = time.time() - ts
    print "time spend", span
    return ret

print calcPi(3000,4000)

IO 密集型

IO 密集型的工作是另一种常见的用例，例如网络 WEB 服务器就是一个例子，每秒钟能解决多少个申请时 WEB 服务器的重要指标。

咱们就以网页读取作为最简略的例子 

from math import hypot
import time
import urllib2

urls = ['http://www.google.com', 'http://www.example.com', 'http://www.python.org']

def test(url):
    return urllib2.urlopen(url).read()

def testIO(nbFutures):
    ts = time.time()
    map(test, urls * nbFutures)

    span = time.time() - ts
    print "time spend", span

testIO(10)

在不同并发库下的代码，因为比拟相似，我就不一一列出。大家能够参考计算密集型中代码做参考。

通过测试咱们能够发现，对于 IO 密集型的工作，应用多线程，或者是多过程都能够无效的进步程序的效率，而应用伪线程性能晋升十分显著，eventlet 比没有并发的状况下，响应工夫从 9 秒进步到 0.03 秒。同时 eventlet/gevent 提供了非阻塞的异步调用模式，十分不便。这里举荐应用线程或者伪线程，因为在响应工夫相似的状况下，线程和伪线程耗费的资源更少。

总结

Python 提供了不同的并发形式，对应于不同的场景，咱们须要抉择不同的形式进行并发。抉择适合的形式，岂但要对该办法的原理有所理解，还应该做一些测试和试验，数据才是你做抉择的最好参考。

正文完
python
发表至： python
2022-01-13
 0
关于开发者:SeaTable-开发者版可以运行-Python-了开发自定义数据处理流程更方便
关于python:记录一下自己找python爬虫工程师实习生岗位的经历
关于python:Python中正则表达式模块详解
关于python:用python如何做自动化运维