爬虫(四)并发编程选择多线程?多进程?协程?异步io?

朴灿烈づ我的快乐病毒、 2022-05-23 07:34 148阅读 0赞

#### python的特点: ####

*  像python这种解释性的语言来讲, 执行效率肯定比起C/C++等底层靠近机器语言的语言来说肯定是差很多的,但是好在python这种语言有着很好的代码可读性, 同样多行的代码, python能干比其他语言要多的多的事情, 所以python来做开发的效率是非常高的, 这对企业这种需要抢占时间, 市场来说是非常重要的.
 *  现在的解释性语言都会使用**JIT技术**, 也叫即时**编译技术**, 就是将要用的代码即时编译成二进制的既期待吗, 加快执行效率, 如果你的程序需要很高的执行效率, 那最好用C语言去写这一段代码, 通过python去调用即可.
 *  如果你的代码需要很高的保密性, 那么也可以使用C语言去书写这一段代码, 因为C语言的代码编译以后被破解的可能性极低, 然后我们再用python去调用就可以了

#### 并发编程 ####

与大多数语言不一样的是python的多线程被全局解释器锁限制了, 所以没办法发挥CPU的多核特性, 无法使CPU满载运行, 所以要实现多任务就要用python的多进程来完成了.

##### 并发编程的选择: #####

*  多线程 + GIL(全局解释器锁) + 共享内存 —>适合I/O密集型任务
 *  多进程 + IPC(socket/pipe)
 *  多进程 + 微线程(协程) —>适合计算密集型任务
 *  单线程 + 异步I/O —>适合I/O密集型任务
 *  node.js —>用js写服务器性能要比java写的要好很多, 因为node.js的工作方式也是异步I/O
 *  相比**异步I/O** , **同步I/O** 对**CPU的利用率很低**, 相比而言性能要低得多!

##### 什么是协程? #####

*  协程是在一个线程中的
 *  协程是一个消费型的生成器
 *  两个函数在互不调用的情况下相互协作

###### 简单的协程 ######

# 生成菲波拉契数的生成器
    def fib():
        a, b = 0, 1
        while True:
            a, b = b, a + b
            yield a
    
    # 计数, 从n倒数到1的生成器
    def countdown(n):
        for i in range(n):
            yield n - i
        # while n > 0:
            # yield n
            # n -= 1
    
    def main():
        # 获取生成器, 生成器不一定要执行完, 因为有yield
        gen = fib()
        for _ in range(20):
            # 预激活生成器
            print(next(gen))
    
    
    if __name__ == '__main__':
        main()  ---> (1,1,2,3,5,8,13,21,34.....)
        c = countdown(100)
        for i in c:
            print(i)  --->(100,99,98.....,3,2,1)

#### 工作类型任务分类 ####

###### 计算密集型 vs IO密集型 ######

1.  使用多进程对I/O密集型任务的性能没有实质性的改善
2.  多进程可以利用CPU的多核特性对计算密集型任务有用
3.  对于网络爬虫这种I/O密集型任务还有一种非常好的处理方式, 就是单线程+异步I/O
4.  所谓的异步I/O就是非阻塞式的I/O操作也就是I/O操作时没有中断CPU
5.  多线程编程最担心什么? - 多个线程竞争资源(并发数据访问)
6.  如果要保证数据的安全性就需要加锁进行保护threading.Lock  
    acquire() / release()
7.  如果多个线程获取锁的方式不正确那么就有可能导致死锁(dead lock)
8.  加锁就意味着线程需要排队等待, 相当于并发代码变成了串行执行.  
    所以在写多线程代码时可以尽量避开对资源的竞争
9.  一种做法就是通过ThreadLocal对线程绑定资源让每个线程持有资源的副本

#### 面向对象多进程爬虫 ####

import logging
    from random import random
    from enum import Enum, unique
    from time import sleep
    from urllib.parse import urlparse
    from os import getpid
    
    import requests
    from bs4 import BeautifulSoup
    from multiprocessing import Process, Pool, Queue
    
    @unique
    class SpiderStatus(Enum):
        """ 枚举类:用于定义符号常量, unique代表值唯一 """
        IDEL = 0
        WORIKING = 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 Exception as e:
                pass
                # logging.error(e)
        return page_html
    
     #包装类,用于重试次数与等待时间, 捕获异常, __call__是在@Retry(*args, **kwargs)的时候自动调用
    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):
    
                try:
                    return fn(*args, **kwargs)
                except self.errors as e:
                    logging.error(e)
                    sleep((random() + 1) * self.wait_secs)
    
            return wrapper
    
     #爬虫类
    class Spider(object):
    
        def __init__(self):
            self.status = SpiderStatus.IDEL
    
        @Retry()
        def fetch(self, current_url, *, user_agent=None, proxies=None, charsets=('utf-8',)):
        # getpid获取当前进程的pid号
            print(f'Process[{getpid()}]:{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=charsets) if resp.status_code == 200 else None
    
        # 解析网页
        def parse(self, html_page, *, domain='m.sohu.com'):
            soup = BeautifulSoup(html_page, 'lxml')
            url_links = []
            for a_tag in soup.body.select('a[href]'):
                parser = urlparse(a_tag.attrs['href'])
                netloc = parser.netloc or 'm.sohu.com'
                scheme = parser.scheme or 'http'
                if netloc == domain and scheme != 'javascript':
                    path = parser.path
                    query = '?' + parser.query if parser.query else ''
                    full_url = f'{scheme}://{netloc}{path}{query}'
                    url_links.append(full_url)
            return url_links
    
        # 摘取
        def extract(self, html_page):
            pass
    
        # 储存
        def store(self, data_dict):
            pass
    
    #爬虫进程类, 继承自Process
    class SpiderProcess(Process):
    
        def __init__(self, spider, tasks_queue):
            super().__init__()
            self.daemon = True
            self.spider = spider
            self.tasks_queue = tasks_queue
    
        # 重写run方法, 创建该类实例后自动调用该函数
        def run(self):
            while True:
                current_url = self.tasks_queue.get()
                self.spider.status = SpiderStatus.WORIKING
                html_page = self.spider.fetch(current_url)
                if html_page not in [None, '']:
                    url_links = self.spider.parse(html_page)
                    for url_link in url_links:
                        self.tasks_queue.put(url_link)
                self.spider.status = SpiderStatus.IDEL
    
     #判断爬虫队列中是否空闲状态
    def is_any_alive(spider_threads):
        return any([spider_thread.spider.status == SpiderStatus.WORIKING for spider_thread in spider_threads])
    
    
    def main():
        # 这是multiprocessing中的Queue类,与queue中的Queue进行区分, 这是多进程的.
        task_queue = Queue()
        task_queue.put('http://m.sohu.com/')
        # 创建多进程的队列
        spider_processes = [SpiderProcess(Spider(), task_queue) for i in range(10)]
    
        # 线程池, 已注释
        # pool = Pool(10)
    
        # 让线程启动
        for spider_process in spider_processes:
            spider_process.start()
    
        while not task_queue.empty() or is_any_alive(spider_processes):
            sleep(30)
            pass
    
        # pool.close()
        print('Over!')
    
    
    if __name__ == '__main__':
        main()