python 多进程

Python异步方案的持续演进之路

 

作者 tonnie 关注

2017.01.14 23:47* 字数 2567 阅读 414评论 4喜欢 9

小梁是一个忠实的电影好爱者，有一天，小梁看到豆瓣这个网站，发现了很多自己喜欢的内容，恰好小梁是个程序猿，于是心血来潮的他决定写个程序，把豆瓣Top250的电影列表给爬下来。小梁平时是个Python发烧友，做起这些事情来自然是得心应手，于是他欣喜地撸起袖子就是干！果不其然，不到十分钟，小梁就写好了第一个程序。

#-*- coding:utf-8 -*-
import urllib.request
import ssl
from lxml import etree

url = 'https://movie.douban.com/top250'
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_1)

def fetch_page(url):
    response = urllib.request.urlopen(url, context=context)
    return response

def parse(url):
    response = fetch_page(url)
    page = response.read()
    html = etree.HTML(page)

    xpath_movie = '//*[@id="content"]/div/div[1]/ol/li'
    xpath_title = './/span[@class="title"]'
    xpath_pages = '//*[@id="content"]/div/div[1]/div[2]/a'

    pages = html.xpath(xpath_pages)
    fetch_list = []
    result = []

    for element_movie in html.xpath(xpath_movie):
        result.append(element_movie)

    for p in pages:
        fetch_list.append(url + p.get('href'))

    for url in fetch_list:
        response = fetch_page(url)
        page = response.read()
        html = etree.HTML(page)
        for element_movie in html.xpath(xpath_movie):
            result.append(element_movie)

    for i, movie in enumerate(result, 1):
        title = movie.find(xpath_title).text         
        print(i, title)


def main():
    parse(url)

if __name__ == '__main__':
    main()

程序也不出意外地正常运行。

但是，这个程序异常的慢，有多慢呢？小梁在主函数中加了下面一段代码。

def main():
    start = time()    
    for i in range(5):
        parse(url)
    end = time()
    print ('Cost {} seconds'.format((end - start) / 5))

发现总共耗时7.6秒！！

小梁不禁陷入了沉思...

小梁突然想起了两天前小张同学给他安利的一个库，叫requests，比那urllib,urllib2,urllib3,urllibn...不知高到哪里去了！小梁兴致勃勃地修改程序，用requests代替了标准库urllib。

import requests
from lxml import etree
from time import time

url = 'https://movie.douban.com/top250'

def fetch_page(url):
    response = requests.get(url)
    return response

def parse(url):
    response = fetch_page(url)
    page = response.content
    html = etree.HTML(page)

    xpath_movie = '//*[@id="content"]/div/div[1]/ol/li'
    xpath_title = './/span[@class="title"]'
    xpath_pages = '//*[@id="content"]/div/div[1]/div[2]/a'

    pages = html.xpath(xpath_pages)
    fetch_list = []
    result = []

    for element_movie in html.xpath(xpath_movie):
        result.append(element_movie)

    for p in pages:
        fetch_list.append(url + p.get('href'))

    for url in fetch_list:
        response = fetch_page(url)
        page = response.content
        html = etree.HTML(page)
        for element_movie in html.xpath(xpath_movie):
            result.append(element_movie)

    for i, movie in enumerate(result, 1):
        title = movie.find(xpath_title).text
        # print(i, title)

结果一测，6.5秒！虽然比用urllib快了1秒多，但是总体来说，他们基本还是处于同一水平线的，程序并没有快很多，这一点的差距是requests对请求做了优化导致的，根据业务场景选择适当的轮子能避免踩许多的坑。

小梁不禁暗想：是我的程序写的太挫了吗？会不会是lxml这个库解析的速度太慢了，用正则表达式会不会好一些？

于是小梁把lxml库换成了标准的re库。

#-*- coding:utf-8 -*-
import requests
from time import time
import re

url = 'https://movie.douban.com/top250'

def fetch_page(url):
    response = requests.get(url)
    return response

def parse(url):
    response = fetch_page(url)
    page = response.content

    fetch_list = set()
    result = []

    for title in re.findall(rb'<a href=.*s.*<span class="title">(.*)</span>', page):
        result.append(title)

    for postfix in re.findall(rb'<a href="(?start=.*?)"', page):
        fetch_list.add(url + postfix.decode())

    for url in fetch_list:
        response = fetch_page(url)
        page = response.content
        for title in re.findall(rb'<a href=.*s.*<span class="title">(.*)</span>', page):
            result.append(title)

    for i, title in enumerate(result, 1):
        title = title.decode()
        # print(i, title)

再一跑，咦，又足足提升了将近一秒！

小梁心里暗爽，程序变得更短了，运行得也更快了，感觉离成功越来越近了，但小梁眉头一皱，很快地意识到了一个问题，这样写出来的程序虽然看起来更短了，但所做的都是在盲目地求快，但完全没有扩展性可言！虽然这样做可以满足普通的需求场景，但当程序逻辑变复杂时，依赖原生正则表达式的程序会更加难以维护！借助一些专门做这些事情的解析库，才能使程序变得清晰。其次，这种网络应用通常瓶颈都在IO层面，解决等待读写的问题比提高文本解析速度来的更有性价比！小梁想起了昨天上操作系统课时老师讲的多进程和多线程概念，正好用他们来解决实际问题。

既然时间都耗在网络IO上了，那我每一页的获取都用一个线程去处理不就好了吗，于是小梁写出了多线程版本的代码。

#-*- coding:utf-8 -*-
import requests
from lxml import etree
from time import time
from threading import Thread

url = 'https://movie.douban.com/top250'

def fetch_page(url):
    response = requests.get(url)
    return response

def parse(url):
    response = fetch_page(url)
    page = response.content
    html = etree.HTML(page)

    xpath_movie = '//*[@id="content"]/div/div[1]/ol/li'
    xpath_title = './/span[@class="title"]'
    xpath_pages = '//*[@id="content"]/div/div[1]/div[2]/a'

    pages = html.xpath(xpath_pages)
    fetch_list = []
    result = []

    for element_movie in html.xpath(xpath_movie):
        result.append(element_movie)

    for p in pages:
        fetch_list.append(url + p.get('href'))

    def fetch_content(url):
        response = fetch_page(url)
        page = response.content
        html = etree.HTML(page)
        for element_movie in html.xpath(xpath_movie):
            result.append(element_movie)

    threads = []
    for url in fetch_list:
        t = Thread(target=fetch_content, args=[url])
        t.start()
        threads.append(t)

    for t in threads:
        t.join()

    for i, movie in enumerate(result, 1):
        title = movie.find(xpath_title).text
        # print(i, title)

效果果然立竿见影！多线程有效的解决了阻塞等待的问题，这个程序足足比之前的程序快了80%！只需要1.4秒就可完成电影列表的抓取。

但小梁还是觉得不够过瘾，既然Python的多线程也受制于GIL，为什么我不用多进程呢？于是话不多说又撸出了一个基于多进程的版本。用4个进程的进程池来并行处理网络数据。

#-*- coding:utf-8 -*-
import requests
from lxml import etree
from time import time
from concurrent.futures import ProcessPoolExecutor

url = 'https://movie.douban.com/top250'

def fetch_page(url):
    response = requests.get(url)
    return response

def fetch_content(url):
    response = fetch_page(url)
    page = response.content
    return page

def parse(url):
    page = fetch_content(url)
    html = etree.HTML(page)

    xpath_movie = '//*[@id="content"]/div/div[1]/ol/li'
    xpath_title = './/span[@class="title"]'
    xpath_pages = '//*[@id="content"]/div/div[1]/div[2]/a'

    pages = html.xpath(xpath_pages)
    fetch_list = []
    result = []

    for element_movie in html.xpath(xpath_movie):
        result.append(element_movie)

    for p in pages:
        fetch_list.append(url + p.get('href'))

    with ProcessPoolExecutor(max_workers=4) as executor:
        for page in executor.map(fetch_content, fetch_list):
            html = etree.HTML(page)
            for element_movie in html.xpath(xpath_movie):
                result.append(element_movie)

    for i, movie in enumerate(result, 1):
        title = movie.find(xpath_title).text
        # print(i, title)

结果是2秒，甚至还不如多线程的版本。

(注：ThreadPoolExecutor和ProcessPoolExecutor是Python3.2之后引入的分别对线程池和进程池的一个封装，如果使用Python2.x，需要安装futures这个库才能使用它们。)

小梁立马就傻眼了，这跟他的预期完全不符合啊。

多进程带来的优点（cpu处理）并没有得到体现，反而创建和调度进程带来的开销要远超出它的正面效应，拖了一把后腿。即便如此，多进程带来的效益相比于之前单进程单线程的模型要好得多。

正当小梁在苦苦思索还有什么方法可以提高性能时，他无意中看到一篇文章，里面提到了协程相比于多进程和多线程的优点（多进程和多线程除了创建的开销大之外还有一个难以根治的缺陷，就是处理进程之间或线程之间的协作问题，因为是依赖多进程和多线程的程序在不加锁的情况下通常是不可控的，而协程则可以完美地解决协作问题，由用户来决定协程之间的调度。），小梁折腾起来也是不甘人后啊，他搜索了一些资料，思考如何用协程来加强自己的程序。

很快，小梁就发现了一个基于协程的网络库，叫做gevent，而且更爽的是，听说用了gevent的猴子补丁后，整个程序就会变成异步的了！

小梁迫不及待地要看看这到底是什么黑科技！马上写出了基于gevent的栗子：

#-*- coding:utf-8 -*-
import requests
from lxml import etree
from time import time
import gevent
from gevent import monkey
monkey.patch_all()

url = 'https://movie.douban.com/top250'

def fetch_page(url):
    response = requests.get(url)
    return response

def fetch_content(url):
    response = fetch_page(url)
    page = response.content
    return page

def parse(url):
    page = fetch_content(url)
    html = etree.HTML(page)

    xpath_movie = '//*[@id="content"]/div/div[1]/ol/li'
    xpath_title = './/span[@class="title"]'
    xpath_pages = '//*[@id="content"]/div/div[1]/div[2]/a'

    pages = html.xpath(xpath_pages)
    fetch_list = []
    result = []

    for element_movie in html.xpath(xpath_movie):
        result.append(element_movie)

    for p in pages:
        fetch_list.append(url + p.get('href'))

    jobs = [gevent.spawn(fetch_content, url) for url in fetch_list]
    gevent.joinall(jobs)
    [job.value for job in jobs]

    for page in [job.value for job in jobs]:
        html = etree.HTML(page)
        for element_movie in html.xpath(xpath_movie):
            result.append(element_movie)

    for i, movie in enumerate(result, 1):
        title = movie.find(xpath_title).text
        # print(i, title)

只有1.2秒，果然很快！而且我们看整个程序，几乎看不到有异步处理的影子，gevent给予了我们一种以同步逻辑来书写异步程序的能力，看monkey.patch_all()这段代码，它是整个程序实现异步的黑科技，当我们给程序打了猴子补丁后，Python程序在运行时会动态地将一些网络库（例如socket，thread）替换掉，变成异步的库。使得程序在进行网络操作的时候都变成异步的方式去工作，效率就自然提升很多了。

虽然程序变得很快了，但小梁整个人都是懵逼的啊，gevent的魔术给他带来了一定的困惑，而且他觉得gevent这玩意实在不好学，跟他心目中Pythonic的清晰优雅还是有距离的。Python社区显然也意识到Python需要一个独立的标准库来支持协程，于是就有了后来的asyncio。

小梁把同步的requests库改成了支持asyncio的aiohttp库，使用3.5的async/await语法（3.5之前用@asyncio.coroutine和yield from代替）写出了协程版本的例子。

#-*- coding:utf-8 -*-
from lxml import etree
from time import time
import asyncio
import aiohttp

url = 'https://movie.douban.com/top250'

async def fetch_content(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()
    return ''

async def parse(url):
    page = await fetch_content(url)
    html = etree.HTML(page)

    xpath_movie = '//*[@id="content"]/div/div[1]/ol/li'
    xpath_title = './/span[@class="title"]'
    xpath_pages = '//*[@id="content"]/div/div[1]/div[2]/a'

    pages = html.xpath(xpath_pages)
    fetch_list = []
    result = []

    for element_movie in html.xpath(xpath_movie):
        result.append(element_movie)

    for p in pages:
        fetch_list.append(url + p.get('href'))

    tasks = [fetch_content(url) for url in fetch_list]
    pages = await asyncio.gather(*tasks)

    for page in pages:
        html = etree.HTML(page)
        for element_movie in html.xpath(xpath_movie):
            result.append(element_movie)

    for i, movie in enumerate(result, 1):
        title = movie.find(xpath_title).text
        # print(i, title)


def main():
    loop = asyncio.get_event_loop()    
    start = time()    
    for i in range(5):
        loop.run_until_complete(parse(url))
    end = time()
    print ('Cost {} seconds'.format((end - start) / 5))
    loop.close()

1.7秒，也不错。用上了async/await语法使得程序的可读性提高了不少。那么这个协程是怎样运作的呢，很容易的理解，总个有10个请求，除了第1个请求是要等待之外（先要抓取分页），剩余9个请求都是协作式的，由Event Loop进行事件调度，试想一下有9个人去仓库拿东西，但要通过咨询仓库管理员来获取货物，仓库里有许多工人，现在如果是同步模型，会是这样的场景：小一，小二，小三...去仓库排队，小一问管理员要货物，管理员通知工人去仓库搬运小一需要的东西，小一就在那里等待，直到取到自己的东西，离开，轮到小二，小二进行同样的操作…而如果是协作模型，则会是这样的场景：小一，小二，小三...去仓库排队，小一问管理员要货物，管理员通知工人去仓库搬运小一需要的东西，管理员对小一说：你的需要等一会儿，去后面等一下，现在轮到小二了。然后小一就排到队伍的最后面，小二问管理员后，管理员对小二说：你的需要等一会儿，去后面等一下，现在轮到小三了，然后小二也排到队伍的最后面，接着每个人都执行一样的操作...直到小九离开，轮到小一了，这时候小A询问管理员他的东西取好没有，这时候又分两种情况：如果取好了，小A领取货物离开，还没取好，小A继续返回到队伍的最后面，以此类推，直至所有人都领取完自己的货物。试想一下，如果每个人平均等待领取货物的时间需要10秒，离开的时间需要1秒，那么同步模型的总耗时最少需要99秒，而协作模型最少只需要同步模型的1/3时间。在这个例子中，小一，小二..小九对应了每个分页的请求，他们都是被aiohttp进行异步封装的，管理员则扮演了Event Loop，类似一个协调者的角色，工人的搬运过程则类似于HTTP协议的传输过程，当然，这些只是比喻，实际上过程要复杂的多。（关于协程我之前的文章已经提过，这里不再阐述）

经过一番洗礼后，小梁对异步有了更加深刻的认识。异步方式有很多，这里列出了比较常见的几种，在实际使用中，应该根据使用场景来挑选最合适的应用方案，影响程序效率的因素有很多，以上不同的异步方式在不同的场景下也会有不一样的表现，不要抱死在一个大树上，该用同步的地方用同步，该用异步的地方异步，这样才能构建出更加灵活的网络应用。

End

---

如果你是小梁，你会有更好的方案吗？欢迎讨论。