Python网络爬虫 第四章 多线程+异步协程

一、多线程抓取北京新发地菜价

多线程、多进程和线程池等的概念，我单独成章了，算到Python基础知识里面，https://www.cnblogs.com/wkfvawl/p/14729542.html

这里就直接开启练习，抓取菜价其实在第二章已经讲过了，那时候用的是bs4解析的网页，这里使用xpath配合多线程。

注意到新发地网站菜价表格网页的url是按照序号递增的，像第一页是

http://www.xinfadi.com.cn/marketanalysis/0/list/1.shtml

第二页是

http://www.xinfadi.com.cn/marketanalysis/0/list/2.shtml

这样，只需要遍历构造url即可得到所有需要的网页链接，但如果是单线程一个个的执行必然效率会很低，那就可以试一试多线程。

使用谷歌浏览器F12的功能，直接获取到表格的xpath。

# 1. 如何提取单个页面的数据
# 2. 上线程池,多个页面同时抓取
import requests
from lxml import etree
import csv
from concurrent.futures import ThreadPoolExecutor

f = open("data.csv", mode="w", encoding="utf-8")
csvwriter = csv.writer(f)


def download_one_page(url):
    # 拿到页面源代码
    resp = requests.get(url)
    html = etree.HTML(resp.text)
    table = html.xpath("/html/body/div[2]/div[4]/div[1]/table")[0]
    # 去掉表头 下面两种方法都想
    # trs = table.xpath("./tr")[1:] # 从第1个开始 去掉第0个表头
    trs = table.xpath("./tr[position()>1]") # 位置大于1
    # 拿到每个tr
    for tr in trs:
        txt = tr.xpath("./td/text()") # tr中找td td中找文本
        # 对数据做简单的处理: \  / 去掉
        txt = (item.replace("\", "").replace("/", "") for item in txt)
        # 把数据存放在文件中
        csvwriter.writerow(txt)
    print(url, "提取完毕!")


if __name__ == '__main__':
    # for i in range(1, 14870):  # 效率及其低下
    #     download_one_page(f"http://www.xinfadi.com.cn/marketanalysis/0/list/{i}.shtml")

    # 创建线程池 50个线程
    with ThreadPoolExecutor(50) as t:
        for i in range(1, 200):  # 199 * 20 = 3980
            # 把下载任务提交给线程池
            t.submit(download_one_page, f"http://www.xinfadi.com.cn/marketanalysis/0/list/{i}.shtml")

    print("全部下载完毕!")

二、协程

协程是并发编程里面很重要的概念，感觉如果要真正弄明白，可能需要完完整整写一章博客，这里就先简单介绍一些基本概念和应用。

协程能够更加⾼效的利⽤CPU，其实, 我们能够⾼效的利⽤多线程来完成爬⾍其实已经很6了。但是，从某种⻆度讲, 线程的执⾏效率真的就⽆敌了么? 我们真的充分的利⽤CPU资源了么? ⾮也~ ⽐如, 我们来看下⾯这个例⼦。我们单独的⽤⼀个线程来完成某⼀个操作，看看它的效率是否真的能把CPU完全利⽤起来。

import time
def func():
 print("我爱黎明")
 time.sleep(3)
 print("我真的爱黎明")
func()

各位请看，在该程序中, 我们的func()实际在执⾏的时候⾄少需要3秒的时间来完成操作，中间的三秒钟需要让我当前的线程处于阻塞状态。阻塞状态的线程 CPU是不会来执⾏的，那么此时cpu很可能会切换到其他程序上去执⾏。此时, 对于你来说, CPU其实并没有为你⼯作(在这三秒内), 那么我们能不能通过某种⼿段, 让CPU⼀直为我⽽⼯作，尽量的不要去管其他⼈。

我们要知道CPU⼀般抛开执⾏周期不谈，如果⼀个线程遇到了IO操作, CPU就会⾃动的切换到其他线程进⾏执⾏. 那么, 如果我想办法让我的线程遇到了IO操作就挂起, 留下的都是运算操作. 那CPU是不是就会⻓时间的来照顾我~.
以此为⽬的, 伟⼤的程序员就发明了⼀个新的执⾏过程. 当线程中遇到了IO操作的时候, 将线程中的任务进⾏切换, 切换成⾮ IO操作. 等原来的IO执⾏完了. 再恢复回原来的任务中。

这里来看一个协程程序

import asyncio
import time

async def func1():
    print("你好啊, 我叫test1")
    time.sleep(3)  # 当程序出现了同步操作的时候. 异步就中断了
    print("你好啊, 我叫test1")


async def func2():
    print("你好啊, 我叫test2")
    time.sleep(2)
    print("你好啊, 我叫test2")


async def func3():
    print("你好啊, 我叫test3")
    time.sleep(4)
    print("你好啊, 我叫test3")


if __name__ == '__main__':
    f1 = func1()
    f2 = func2()
    f3 = func3()
    # 任务列表
    tasks = [
        f1, f2, f3
    ]
    t1 = time.time()
    # 一次性启动多个任务(协程)
    asyncio.run(asyncio.wait(tasks))
    t2 = time.time()
    print(t2 - t1)

 运行的结果并没有如同协程定义那样，产生异步效果，反而是同步的？这是因为里面的time.sleep()是同步操作，导致异步中断了，正确的写法应该是这样：

import asyncio
import time

async def func1():
    print("你好啊, 我叫test1")
    await asyncio.sleep(3)  # 异步操作的代码 await挂起
    print("你好啊, 我叫test1")

async def func2():
    print("你好啊, 我叫test2")
    await asyncio.sleep(2)
    print("你好啊, 我叫test2")

async def func3():
    print("你好啊, 我叫test3")
    await asyncio.sleep(4)
    print("你好啊, 我叫test3")

async def main():
    # 第一种写法
    # f1 = func1()
    # await f1  # 一般await挂起操作放在协程对象前面
    # 第二种写法(推荐)
    # tasks = [
    #     func1(),
    #     func2(),
    #     func3()
    # ]
    tasks = [
        asyncio.create_task(func1()),  # py3.8以后加上asyncio.create_task()
        asyncio.create_task(func2()),
        asyncio.create_task(func3())
    ]
    await asyncio.wait(tasks)


if __name__ == '__main__':
    t1 = time.time()
    # 一次性启动多个任务(协程)
    asyncio.run(main())
    t2 = time.time()
    print(t2 - t1)

 从程序运行时间上来看利用异步协程直接从9秒减少到了4秒。这里需要asyncio的支持。

关于asyncio的介绍参考https://www.liaoxuefeng.com/wiki/1016959663602400/1017970488768640

await关键词。异步io的关键在于，await io操作，此时，当前携程就会被挂起，时间循环转而执行其他携程，但是要注意前面这句话，并不是说所有携程里的await都会导致当前携程的挂起，要看await后面跟的是什么，如果跟的是我们定义的携程，则会执行这个携程，如果是asyncio模块制作者定义的固有携程，比如模拟io操作的asyncio.sleep,以及io操作，比如网络io:asyncio.open_connection这些，才会挂起当前携程。

三、aiohttp模块应用

前面我们使用asyncio来实现了异步协程，那我们该如何将异步协程应用到爬虫上呢？其实爬虫在连接到要爬取的网页上的过程，也是一个类似IO的过程，这里介绍一下aiohttp，是一个用于asyncio和Python的异步HTTP客户端/服务器。

以第二章讲过的唯美壁纸网站为例。之前同步时候用的requests ，换成了异步操作的aiohttp。

import asyncio
import aiohttp

urls = [
    "http://kr.shanghai-jiuxin.com/file/2020/1031/191468637cab2f0206f7d1d9b175ac81.jpg",
    "http://kr.shanghai-jiuxin.com/file/2020/1031/563337d07af599a9ea64e620729f367e.jpg",
    "http://kr.shanghai-jiuxin.com/file/2020/1031/774218be86d832f359637ab120eba52d.jpg"
]

async def aiodownload(url):
    # 发送请求.
    # 得到图片内容
    # 保存到文件
    name = url.rsplit("/", 1)[1]  # 从右边切, 切一次. 得到[1]位置的内容
    # 加with 上下文管理器
    # s = aiohttp.ClientSession() <==> requests.session()
    async with aiohttp.ClientSession() as session:  # requests
        async with session.get(url) as resp:  # resp = requests.get()
            # 请求回来了. 写入文件
            # 可以自己去学习一个模块, aiofiles
            with open(name, mode="wb") as f:  # 创建文件
                f.write(await resp.content.read())  # 读取内容是异步的. 需要await挂起, resp.text()

    print(name, "搞定")

async def main():
    # tasks列表
    tasks = []
    for url in urls:
        tasks.append(aiodownload(url))
    await asyncio.wait(tasks)

if __name__ == '__main__':
    asyncio.run(main())

这个程序还有待改进空间的，创建文件写文件也是一个IO操作，也是可以异步的，要引入aiofiles这个后面会讲。

四、利用协程下载小说

这次我们下载百度小说上的《西游记》。http://dushu.baidu.com/pc/detail?gid=4306063500

 F12抓包，找到了每一章节的名称和cid

http://dushu.baidu.com/api/pc/getCatalog?data={"book_id":"4306063500"}

经历了之前的实践，是不是感觉这次的url优点奇怪？date后面是一个json?

接着为了获取每个章节里面的内容，点开一章，发现内容存在于http://dushu.baidu.com/api/pc/getChapterContent?data={"book_id":"4306063500","cid":"4306063500|11348571","need_bookinfo":1}中

通过更换cid我们就能很轻松的获取到其他章节的内容了。

 在编写程序之前，先要清楚我们需要做什么工作？

其实这是一个同步异步相结合的工作

• 1. 同步操作: 访问getCatalog 拿到所有章节的cid和名称
• 2. 异步操作: 访问getChapterContent 下载所有的文章内容

import requests
import asyncio
import aiohttp
import aiofiles
import json

async def aiodownload(cid, b_id, title):
    data = {
        "book_id": b_id,
        "cid": f"{b_id}|{cid}",
        "need_bookinfo": 1
    }
    # 转成json
    data = json.dumps(data)
    url = f"http://dushu.baidu.com/api/pc/getChapterContent?data={data}"

    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            dic = await resp.json()

            async with aiofiles.open(title, mode="w", encoding="utf-8") as f:
                await f.write(dic['data']['novel']['content'])  # 把小说内容写出


async def getCatalog(url):
    resp = requests.get(url)
    # 取json
    dic = resp.json()
    tasks = []
    for item in dic['data']['novel']['items']:  # item就是对应每一个章节的名称和cid
        title = './novel/' + item['title'] + '.txt'
        cid = item['cid']
        # 准备异步任务
        tasks.append(aiodownload(cid, b_id, title))
    await asyncio.wait(tasks)


if __name__ == '__main__':
    b_id = "4306063500"
    url = 'http://dushu.baidu.com/api/pc/getCatalog?data={"book_id":"' + b_id + '"}'
    asyncio.run(getCatalog(url))

爬虫程序运行速度极快！