爬虫(五)了解协程工作原理/回顾多线程与多进程

小咪咪 2022-05-23 08:46 149阅读 0赞

#### 多线程和多进程回顾 ####

*  在前面的<进程和线程>一文中, 我们已经对在Python中使用多进程和多线程实现并发编程进行了简单的讲解, 在此我们补充几个知识点.

###### threading.local类 ######

*  使用线程时最不愿意遇到的情况就是多个线程竞争资源, 在这种情况下为了保证资源状态的正确性, 我们可能需要对资源进行加锁保护的处理, 这一方面会导致程序失去并发性, 另外如果多个线程竞争多个资源时, 还有可能因为加锁方式的不当导致死锁. 要解决多个线程竞争资源的问题, 其中一个方案就是让每个线程都持有资源的副本(拷贝), 这样每个线程可以操作自己所持有的资源, 从而规避对资源的竞争.
 *  要实现将资源和持有资源的线程进行绑定的操作, 最简单的做法就是使用threading模块和local类, 在网络爬虫开发中, 就可以使用local类为每个线程绑定一个MySQL数据库连接或Redis客户端对象, 这样通过线程可以直接获得这些资源, 既解决了资源竞争的问题, 又避免了再函数和方法调用时传递这些资源. 具体的请参考 多线程爬取’手机搜狐网’(Redis版本)的代码.

###### concurrent.futures模块 ######

*  Python3.2带来了`concurrent.futures`模块, 这个模块包含了线程池和进程池, 管理并行编程任务, 处理非确定性的执行流程, 进程/线程同步等功能. 关于这部分的内容推荐大家阅读[《Python并行编程》][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 : 引入`async`和`await`, 可以使用`async def`来定义一个协程函数, 这个函数中不能包含任何形式的`yield`语句, 但是可以使用`return`或`await`从协程中返回值.

##### 示例代码 #####

*  生成器叠加生成器

# 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()) # 在fib这个生成器上叠加一个even生成器, 通过循环取出数据的结果是叠加后的效果, 即从10个fib数列中取出为偶数的元素.
        for _ in range(10):
            print(next(gen))
    
    
    if __name__ == '__main__':
        main()

*  协程 - 数据的消费者

from time import sleep
    
    
    # 生成器 - 数据生产者
    def countdown_gen(n, consumer):
        # 通过send(None)或者next(consumer)会激活消费者生成器
        consumer.send(None)
        # 当n>0的时候, 向消费者发送数据n
        while n > 0:
            consumer.send(n)
            n -= 1
        # 当循环结束的时候, 发送None, 表示数据消耗完毕
        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_con里面有yield 所以程序被视作一个生成器, 生成器函数不一定要执行完, 用到的时候才会执行,不会阻塞整个程序的进行, 
        # 所以执行countdown_con()会产生一个`GEN_CREATED`的状态的生成器, 等待countdown_gen生产者去激活他变成挂起状态, 即`GEN_SUSPENDED`, 当程序结束的时候, 状态会变成`GEN_CLOSED`
        # 查看状态用print(getgeneratorstate(countdown_con)),获得该生成器的状态
        countdown_gen(5, countdown_con())
    
    
    if __name__ == '__main__':
        main()

> 说明：上面代码中countdown\_gen函数中的第1行consumer.send(None)是为了激活生成器，通俗的说就是让生成器执行到有yield关键字的地方挂起，当然也可以通过next(consumer)来达到同样的效果。如果不愿意每次都用这样的代码来“预激”生成器，可以写一个包装器来完成该操作，代码如下所示。

*  下面代码可以使用`@coroutine`包装器函数对协程进行预激活操作, 不需要再写重复的代码来激活协程.

from functools import wraps
    
    
    def coroutine(fn):
        # @wraps(fn)可以使在调用函数.__name__的时候可以得到函数自身的名称, 而不是包装器名称, 即coroutine的名称
        @wraps(fn)
        def wrapper(*args, **kwargs):
            gen = fn(*args, **kwargs)
            next(gen)
            return gen
    
        return wrapper

*  异步I/O - 非阻塞式I/O操作

但是只有协程还不够，还不足以实现异步IO,我们必须实现消息循环和状态的控制.这就要引入asynico，它直接内置了对异步IO的支持.asyncio的编程模型就是一个消息循环。我们从asyncio模块中直接获取一个EventLoop的引用，然后把需要执行的协程扔到EventLoop中执行，就实现了异步IO。

import asyncio
    
    
    @asyncio.coroutine
    def countdown(name, n):
        while n > 0:
            print(f'Countdown[{name}]: {n}')
            yield from asyncio.sleep(1)
            n -= 1
    
    @asyncio.coroutine
    def countdown1(name, n):
        while n > 0:
            print(f'Countdown[{name}]: {n}')
            yield from asyncio.sleep(2)
            n -= 2
    
    def main():
        # 获取evet_loop的引用
        loop = asyncio.get_event_loop()
        # 创建迭代器队列
        tasks = [
            countdown("A", 10), countdown1("B", 20), countdown('C', 5), countdown1('D', 16)
        ]
        # 将迭代器队列放入asyncio.wait(tasks)中
        loop.run_until_complete(asyncio.wait(tasks))
        # 运行结束, 关闭loop
        loop.close()
    
    
    if __name__ == '__main__':
        # 从运行结果上看, 这4个程序都是并发一起运行的, 就像是几个线程一起运行的结果, 而且因为只有一个线程, 不会产生线程并发带来的影响
        # 这得益于异步I/O, 对于I/O密集型任务, 协程是很好的选择
        main()

*  `async` 和 `await`

import asyncio
    import aiohttp
    
    #请注意，async和await是针对coroutine的新语法，要使用新的语法，只需要做两步简单的替换：
    
    #1. 把@asyncio.coroutine替换为async；
    #2. 把yield from替换为await。
    
    async def download(url):
        print('Fetch:', url)
        # with xxx() 创建上下文环境, 运行结束自动关闭运行的程序
        # 异步创建一个aiohttp.ClientSession的实例, 别名为session
        async with aiohttp.ClientSession() as session:
            # 异步请求url, 返回的结果别名为resp
            async with session.get(url) as resp:
                # 打印 响应(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服务器的功能，对异步操作提供了非常好的支持，有兴趣可以阅读它的[官方文档][Link 1]。

#### 实例 - 多线程爬取“手机搜狐网”所有页面(绑定资源, 使用threading.local类)X ####

import pickle # 导入序列化包
    import zlib # 导入压缩工具包
    from enum import Enum, unique #导入枚举类, 唯一标识
    from hashlib import sha1 # 导入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):
            pass
    
        print('Over!')
    
    
    if __name__ == '__main__':
        main()

[Python]: http://python-parallel-programmning-cookbook.readthedocs.io/zh_CN/latest/index.html
[AIOHTTP]: https://github.com/aio-libs/aiohttp
[Link 1]: https://aiohttp.readthedocs.io/en/stable/