Python-100-Days/Day61-65/64.并发下载.md

10 KiB
Raw Blame History

并发下载

多线程和多进程补充知识点

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()操作来转移执行权,协程之间不是调用者与被调用者的关系。

协程的优势在于以下两点:

  1. 执行效率极高,因为子程序(函数)切换不是线程切换,由程序自身控制,没有切换线程的开销。
  2. 不需要多线程的锁机制,因为只有一个线程,也不存在竞争资源的问题,当然也就不需要对资源加锁保护,因此执行效率高很多。

说明协程适合处理的是I/O密集型任务处理CPU密集型任务并不是它擅长的如果要提升CPU的利用率可以考虑“多进程+多线程”或者“多进程+协程”的工作模式。

历史回顾

  1. Python 2.2第一次提出了生成器最初称之为迭代器的概念PEP 255
  2. Python 2.5:引入了将对象发送回暂停了的生成器这一特性即生成器的send()方法PEP 342
  3. Python 3.3:添加了yield from特性,允许从迭代器中返回任何值(注意生成器本身也是迭代器),这样我们就可以串联生成器并且重构出更好的生成器。
  4. Python 3.4:引入asyncio.coroutine装饰器用来标记作为协程的函数,协程函数和asyncio及其事件循环一起使用来实现异步I/O操作。
  5. Python 3.5:引入了asyncawait,可以使用async def来定义一个协程函数,这个函数中不能包含任何形式的yield语句,但是可以使用returnawait从协程中返回值。

协程实现了协作式并发通过提高CPU的利用率来达到改善性能的目的。著名的三方库aiohttp就是通过协程的方式实现了HTTP客户端和HTTP服务器的功能较之requests有更好的获取数据的性能,有兴趣可以阅读它的官方文档

import asyncio
import aiohttp


async def download(url):
    print('Fetch:', url)
    async with aiohttp.ClientSession() as session:
        async with session.get(url, ssl=False) as resp:
            print(url, '--->', resp.status)
            print(url, '--->', resp.headers)
            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/',
        'http://jd.com/'
    ]
    tasks = [download(url) for url in urls]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()


if __name__ == '__main__':
    main()

实例 - 多线程爬取“手机搜狐网”所有页面

下面我们把之间讲的所有知识结合起来,用面向对象的方式实现一个爬取“手机搜狐网”的多线程爬虫。

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()