併發下載

多線程和多進程回顧

在前面的《進程和線程》一文中，我們已經對在Python中使用多進程和多線程實現併發編程進行了簡明的講解，在此我們補充幾個知識點。

threading.local類

使用線程時最不願意遇到的情況就是多個線程競爭資源，在這種情況下爲了保證資源狀態的正確性，我們可能需要對資源進行加鎖保護的處理，這一方面會導致程序失去併發性，另外如果多個線程競爭多個資源時，還有可能因爲加鎖方式的不當導致死鎖。要解決多個線程競爭資源的問題，其中一個方案就是讓每個線程都持有資源的副本（拷貝），這樣每個線程可以操作自己所持有的資源，從而規避對資源的競爭。

要實現將資源和持有資源的線程進行綁定的操作，最簡單的做法就是使用threading模塊的local類，在網絡爬蟲開發中，就可以使用local類爲每個線程綁定一個MySQL數據庫連接或Redis客戶端對象，這樣通過線程可以直接獲得這些資源，既解決了資源競爭的問題，又避免了在函數和方法調用時傳遞這些資源。具體的請參考本章多線程爬取“手機搜狐網”（Redis版）的實例代碼。

concurrent.futures模塊

Python3.2帶來了concurrent.futures 模塊，這個模塊包含了線程池和進程池、管理並行編程任務、處理非確定性的執行流程、進程/線程同步等功能。關於這部分的內容推薦大家閱讀《Python並行編程》。

分佈式進程

使用多進程的時候，可以將進程部署在多個主機節點上，Python的multiprocessing模塊不但支持多進程，其中managers子模塊還支持把多進程部署到多個節點上。當然，要部署分佈式進程，首先需要一個服務進程作爲調度者，進程之間通過網絡進行通信來實現對進程的控制和調度，由於managers模塊已經對這些做出了很好的封裝，因此在無需瞭解網絡通信細節的前提下，就可以編寫分佈式多進程應用。具體的請參照本章分佈式多進程爬取“手機搜狐網”的實例代碼。

協程和異步I/O

協程的概念

協程（coroutine）通常又稱之爲微線程或纖程，它是相互協作的一組子程序（函數）。所謂相互協作指的是在執行函數A時，可以隨時中斷去執行函數B，然後又中斷繼續執行函數A。注意，這一過程並不是函數調用（因爲沒有調用語句），整個過程看似像多線程，然而協程只有一個線程執行。協程通過yield關鍵字和 send()操作來轉移執行權，協程之間不是調用者與被調用者的關係。

協程的優勢在於以下兩點：

執行效率極高，因爲子程序（函數）切換不是線程切換，由程序自身控制，沒有切換線程的開銷。
不需要多線程的鎖機制，因爲只有一個線程，也不存在競爭資源的問題，當然也就不需要對資源加鎖保護，因此執行效率高很多。

說明：協程適合處理的是I/O密集型任務，處理CPU密集型任務並不是它的長處，如果要提升CPU的利用率可以考慮“多進程+協程”的模式。

歷史回顧

Python 2.2：第一次提出了生成器（最初稱之爲迭代器）的概念（PEP 255）。
Python 2.5：引入了將對象發送回暫停了的生成器這一特性即生成器的send()方法（PEP 342）。
Python 3.3：添加了yield from特性，允許從迭代器中返回任何值（注意生成器本身也是迭代器），這樣我們就可以串聯生成器並且重構出更好的生成器。
Python 3.4：引入asyncio.coroutine裝飾器用來標記作爲協程的函數，協程函數和asyncio及其事件循環一起使用，來實現異步I/O操作。
Python 3.5：引入了async和await，可以使用async def來定義一個協程函數，這個函數中不能包含任何形式的yield語句，但是可以使用return或await從協程中返回值。

示例代碼

生成器 - 數據的生產者。

from time import sleep


# 倒計數生成器
def countdown(n):
    while n > 0:
        yield n
        n -= 1


def main():
    for num in countdown(5):
        print(f'Countdown: {num}')
        sleep(1)
    print('Countdown Over!')


if __name__ == '__main__':
    main()

生成器還可以疊加來組成生成器管道，代碼如下所示。

# Fibonacci數生成器
def fib():
    a, b = 0, 1
    while True:
        a, b = b, a + b
        yield a


# 偶數生成器
def even(gen):
    for val in gen:
        if val % 2 == 0:
            yield val


def main():
    gen = even(fib())
    for _ in range(10):
        print(next(gen))


if __name__ == '__main__':
    main()

協程 - 數據的消費者。

from time import sleep


# 生成器 - 數據生產者
def countdown_gen(n, consumer):
    consumer.send(None)
    while n > 0:
        consumer.send(n)
        n -= 1
    consumer.send(None)


# 協程 - 數據消費者
def countdown_con():
    while True:
        n = yield
        if n:
            print(f'Countdown {n}')
            sleep(1)
        else:
            print('Countdown Over!')


def main():
    countdown_gen(5, countdown_con())


if __name__ == '__main__':
    main()

說明：上面代碼中countdown_gen函數中的第1行consumer.send(None)是爲了激活生成器，通俗的說就是讓生成器執行到有yield關鍵字的地方掛起，當然也可以通過next(consumer)來達到同樣的效果。如果不願意每次都用這樣的代碼來“預激”生成器，可以寫一個包裝器來完成該操作，代碼如下所示。

from functools import wraps


def coroutine(fn):

    @wraps(fn)
    def wrapper(*args, **kwargs):
        gen = fn(*args, **kwargs)
        next(gen)
        return gen

    return wrapper

這樣就可以使用@coroutine裝飾器對協程進行預激操作，不需要再寫重複代碼來激活協程。

異步I/O - 非阻塞式I/O操作。

import asyncio


@asyncio.coroutine
def countdown(name, n):
    while n > 0:
        print(f'Countdown[{name}]: {n}')
        yield from asyncio.sleep(1)
        n -= 1


def main():
    loop = asyncio.get_event_loop()
    tasks = [
        countdown("A", 10), countdown("B", 5),
    ]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()


if __name__ == '__main__':
    main()

async和await。

import asyncio
import aiohttp


async def download(url):
    print('Fetch:', url)
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            print(url, '--->', resp.status)
            print(url, '--->', resp.cookies)
            print('\n\n', await resp.text())


def main():
    loop = asyncio.get_event_loop()
    urls = [
        'https://www.baidu.com',
        'http://www.sohu.com/',
        'http://www.sina.com.cn/',
        'https://www.taobao.com/',
        'https://www.jd.com/'
    ]
    tasks = [download(url) for url in urls]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()


if __name__ == '__main__':
    main()

上面的代碼使用了AIOHTTP這個非常著名的第三方庫，它實現了HTTP客戶端和HTTP服務器的功能，對異步操作提供了非常好的支持，有興趣可以閱讀它的官方文檔。

實例 - 多線程爬取“手機搜狐網”所有頁面

下面我們把之間講的所有知識結合起來，用面向對象的方式實現一個爬取“手機搜狐網”的多線程爬蟲。

import pickle
import zlib
from enum import Enum, unique
from hashlib import sha1
from random import random
from threading import Thread, current_thread, local
from time import sleep
from urllib.parse import urlparse

import pymongo
import redis
import requests
from bs4 import BeautifulSoup
from bson import Binary


@unique
class SpiderStatus(Enum):
    IDLE = 0
    WORKING = 1


def decode_page(page_bytes, charsets=('utf-8',)):
    page_html = None
    for charset in charsets:
        try:
            page_html = page_bytes.decode(charset)
            break
        except UnicodeDecodeError:
            pass
    return page_html


class Retry(object):

    def __init__(self, *, retry_times=3,
                 wait_secs=5, errors=(Exception, )):
        self.retry_times = retry_times
        self.wait_secs = wait_secs
        self.errors = errors

    def __call__(self, fn):

        def wrapper(*args, **kwargs):
            for _ in range(self.retry_times):
                try:
                    return fn(*args, **kwargs)
                except self.errors as e:
                    print(e)
                    sleep((random() + 1) * self.wait_secs)
            return None

        return wrapper


class Spider(object):

    def __init__(self):
        self.status = SpiderStatus.IDLE

    @Retry()
    def fetch(self, current_url, *, charsets=('utf-8', ),
              user_agent=None, proxies=None):
        thread_name = current_thread().name
        print(f'[{thread_name}]: {current_url}')
        headers = {'user-agent': user_agent} if user_agent else {}
        resp = requests.get(current_url,
                            headers=headers, proxies=proxies)
        return decode_page(resp.content, charsets) \
            if resp.status_code == 200 else None

    def parse(self, html_page, *, domain='m.sohu.com'):
        soup = BeautifulSoup(html_page, 'lxml')
        for a_tag in soup.body.select('a[href]'):
            parser = urlparse(a_tag.attrs['href'])
            scheme = parser.scheme or 'http'
            netloc = parser.netloc or domain
            if scheme != 'javascript' and netloc == domain:
                path = parser.path
                query = '?' + parser.query if parser.query else ''
                full_url = f'{scheme}://{netloc}{path}{query}'
                redis_client = thread_local.redis_client
                if not redis_client.sismember('visited_urls', full_url):
                    redis_client.rpush('m_sohu_task', full_url)

    def extract(self, html_page):
        pass

    def store(self, data_dict):
        # redis_client = thread_local.redis_client
        # mongo_db = thread_local.mongo_db
        pass


class SpiderThread(Thread):

    def __init__(self, name, spider):
        super().__init__(name=name, daemon=True)
        self.spider = spider

    def run(self):
        redis_client = redis.Redis(host='1.2.3.4', port=6379, password='1qaz2wsx')
        mongo_client = pymongo.MongoClient(host='1.2.3.4', port=27017)
        thread_local.redis_client = redis_client
        thread_local.mongo_db = mongo_client.msohu 
        while True:
            current_url = redis_client.lpop('m_sohu_task')
            while not current_url:
                current_url = redis_client.lpop('m_sohu_task')
            self.spider.status = SpiderStatus.WORKING
            current_url = current_url.decode('utf-8')
            if not redis_client.sismember('visited_urls', current_url):
                redis_client.sadd('visited_urls', current_url)
                html_page = self.spider.fetch(current_url)
                if html_page not in [None, '']:
                    hasher = hasher_proto.copy()
                    hasher.update(current_url.encode('utf-8'))
                    doc_id = hasher.hexdigest()
                    sohu_data_coll = mongo_client.msohu.webpages
                    if not sohu_data_coll.find_one({'_id': doc_id}):
                        sohu_data_coll.insert_one({
                            '_id': doc_id,
                            'url': current_url,
                            'page': Binary(zlib.compress(pickle.dumps(html_page)))
                        })
                    self.spider.parse(html_page)
            self.spider.status = SpiderStatus.IDLE


def is_any_alive(spider_threads):
    return any([spider_thread.spider.status == SpiderStatus.WORKING
                for spider_thread in spider_threads])


thread_local = local()
hasher_proto = sha1()


def main():
    redis_client = redis.Redis(host='1.2.3.4', port=6379, password='1qaz2wsx')
    if not redis_client.exists('m_sohu_task'):
        redis_client.rpush('m_sohu_task', 'http://m.sohu.com/')

    spider_threads = [SpiderThread('thread-%d' % i, Spider())
                      for i in range(10)]
    for spider_thread in spider_threads:
        spider_thread.start()

    while redis_client.exists('m_sohu_task') or is_any_alive(spider_threads):
        sleep(5)

    print('Over!')


if __name__ == '__main__':
    main()

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

69.併發下載.md

69.併發下載.md

併發下載

多線程和多進程回顧

threading.local類

concurrent.futures模塊

分佈式進程

協程和異步I/O

協程的概念

歷史回顧

示例代碼

實例 - 多線程爬取“手機搜狐網”所有頁面

Files

69.併發下載.md

Latest commit

History

69.併發下載.md

File metadata and controls

併發下載

多線程和多進程回顧

threading.local類

concurrent.futures模塊

分佈式進程

協程和異步I/O

協程的概念

歷史回顧

示例代碼

實例 - 多線程爬取“手機搜狐網”所有頁面