python用装饰器实现多线程

Tomorrow.py

无意中发现github上这么个库tomorrow.py, Magic decorator syntax for asynchronous code in Python 2.7,使用装饰器就可以实现多线程,看起来很神奇。示例如下:

urls = [
    'http://google.com',
    'http://facebook.com',
    'http://youtube.com',
    'http://baidu.com',
    'http://yahoo.com',
]

import time
import requests

def download(url):
    return requests.get(url)

if __name__ == "__main__":
    start = time.time()
    responses = [download(url) for url in urls]
    html = [response.text for response in responses]
    end = time.time()
    print "Time: %f seconds" % (end - start)

这是传统的方式,用tomorrow可以怎么做呢,先pip install tomorrow,注意到会安装依赖concurrent库。这个库在python3.2版本中引入到了标准库,在2.7版本中注意到安装tomorrow的时候会先安装concurrent库。下边是多线程版本,仅仅多了一行:

import time
import requests

from tomorrow import threads

@threads(5)
def download(url):
    return requests.get(url)

if __name__ == "__main__":
    start = time.time()
    responses = [download(url) for url in urls]
    html = [response.text for response in responses]
    end = time.time()
    print "Time: %f seconds" % (end - start)

只是加了一个装饰器@thread(5),就可以多线程执行requests函数了,非常方便,还用个毛线程池。来看一下tomorrow.py源代码,一共才不到50行,其源码如下:

from functools import wraps

from concurrent.futures import ThreadPoolExecutor


class Tomorrow():

    def __init__(self, future, timeout):
        self._future = future
        self._timeout = timeout

    def __getattr__(self, name):
        result = self._wait()
        return result.__getattribute__(name)

    def _wait(self):
        return self._future.result(self._timeout)


def async(n, base_type, timeout=None):
    def decorator(f):
        if isinstance(n, int):
            pool = base_type(n)
        elif isinstance(n, base_type):
            pool = n
        else:
            raise TypeError(
                "Invalid type: %s"
                % type(base_type)
            )
        @wraps(f)
        def wrapped(*args, **kwargs):
            return Tomorrow(
                pool.submit(f, *args, **kwargs),  # 创建future对象
                timeout=timeout
            )
        return wrapped
    return decorator


def threads(n, timeout=None):
    return async(n, ThreadPoolExecutor, timeout)

实际上可以看出就是以下官方示例的一个装饰器封装,原始的版本是这么写的(稍微改了python文档上的示例):

import requests
import concurrent.futures

URLS = ['http://www.baidu.com'] * 10


def load_url(url, timeout):
    return requests.get(url, timeout=timeout).content

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            data = future.result()
        except Exception as exc:
            print('%r generated an exception: %s' % (url, exc))
        else:
            print('%r page is %d bytes' % (url, len(data)))

暂时这个用得不多,以后有时间再慢慢研究下吧,不清楚工程上是不是有直接应用了,不过写个小脚本可以用用。

几个参考:

current.futures

使用python的并发库concurrent.futures实现异步

简单实现并发:python concurrent模块

Vim python-mode. PyLint, Rope, Pydoc, breakpoints from box. 最后推荐这个vim插件,叫做python-mode,集成了众多python插件,包括语法检查,补全等,非常方便,直接相当于一个python的IDE了。