在做爬虫的时候，我们往往可能这些情况：

• 网站比较复杂，会碰到很多重复请求。
• 有时候爬虫意外中断了，但我们没有保存爬取状态，再次运行就需要重新爬取。

还有诸如此类的问题。

那怎么解决这些重复爬取的问题呢？大家很可能都想到了“缓存”，也就是说，爬取过一遍就直接跳过爬取。

那一般怎么做呢？

比如我写一个逻辑，把已经爬取过的 URL 保存到文件或者数据库里面，每次爬取之前检查一下是不是在列表或数据库里面就好了。

是的，这个思路没问题，但有没有想过这些问题：

• 写入到文件或者数据库可能是永久性的，如果我想控制缓存的有效时间，那就还得有个过期时间控制。
• 这个缓存根据什么来判断？如果仅仅是 URL 本身够吗？还有 Request Method、Request Headers 呢，如果它们不一样了，那还要不要用缓存？
• 如果我们有好多项目，难道都没有一个通用的解决方案吗？

的确是些问题，实现起来确实要考虑很多问题。

不过不用担心，今天给大家介绍一个神器，可以帮助我们通通解决如上的问题。

介绍

它就是 requests-cache，是 requests 库的一个扩展包，利用它我们可以非常方便地实现请求的缓存，直接得到对应的爬取结果。

• GitHub：https://github.com/reclosedev/requests-cache
• PyPi：https://pypi.org/project/requests-cache/
• 官方文档：https://requests-cache.readthedocs.io/en/stable/index.html

下面我们来介绍下它的使用。

安装

安装非常简单，使用 pip3 即可：

1

pip3 install requests-cache

安装完毕之后我们来了解下它的基本用法。

基本用法

下面我们首先来看一个基础实例：

1
2
3
4
5
6
7
8
9
10

import requests
import time

start = time.time()
session = requests.Session()
for i in range(10):
    session.get('http://httpbin.org/delay/1')
    print(f'Finished {i + 1} requests')
end = time.time()
print('Cost time', end - start)

这里我们请求了一个网站，是 http://httpbin.org/delay/1，这个网站模拟了一秒延迟，也就是请求之后它会在 1 秒之后才会返回响应。

这里请求了 10 次，那就至少得需要 10 秒才能完全运行完毕。

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11

Finished 1 requests
Finished 2 requests
Finished 3 requests
Finished 4 requests
Finished 5 requests
Finished 6 requests
Finished 7 requests
Finished 8 requests
Finished 9 requests
Finished 10 requests
Cost time 13.17966604232788

可以看到，这里一共用了 13 秒。

那如果我们用上 requests-cache 呢？结果会怎样？

代码改写如下：

1
2
3
4
5
6
7
8
9
10
11

import requests_cache
import time

start = time.time()
session = requests_cache.CachedSession('demo_cache')

for i in range(10):
    session.get('http://httpbin.org/delay/1')
    print(f'Finished {i + 1} requests')
end = time.time()
print('Cost time', end - start)

这里我们声明了一个 CachedSession，将原本的 Session 对象进行了替换，还是请求了 10 次。

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11

Finished 1 requests
Finished 2 requests
Finished 3 requests
Finished 4 requests
Finished 5 requests
Finished 6 requests
Finished 7 requests
Finished 8 requests
Finished 9 requests
Finished 10 requests
Cost time 1.6248838901519775

可以看到，一秒多就爬取完毕了！

发生了什么？

这时候我们可以发现，在本地生成了一个 demo_cache.sqlite 的数据库。

我们打开之后可以发现里面有个 responses 表，里面多了一个 key-value 记录，如图所示：

我们可以可以看到，这个 key-value 记录中的 key 是一个 hash 值，value 是一个 Blob 对象，里面的内容就是 Response 的结果。

可以猜到，每次请求都会有一个对应的 key 生成，然后 requests-cache 把对应的结果存储到了 SQLite 数据库中了，后续的请求和第一次请求的 URL 是一样的，经过一些计算它们的 key 也都是一样的，所以后续 2-10 请求就立马返回了。

是的，利用这个机制，我们就可以跳过很多重复请求了，大大节省爬取时间。

Patch 写法

但是，刚才我们在写的时候把 requests 的 session 对象直接替换了。有没有别的写法呢？比如我不影响当前代码，只在代码前面加几行初始化代码就完成 requests-cache 的配置呢？

当然是可以的，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13

import time
import requests
import requests_cache

requests_cache.install_cache('demo_cache')

start = time.time()
session = requests.Session()
for i in range(10):
    session.get('http://httpbin.org/delay/1')
    print(f'Finished {i + 1} requests')
end = time.time()
print('Cost time', end - start)

这次我们直接调用了 requests-cache 库的 install_cache 方法就好了，其他的 requests 的 Session 照常使用即可。

我们再运行一遍：

1
2
3
4
5
6
7
8
9
10
11

Finished 1 requests
Finished 2 requests
Finished 3 requests
Finished 4 requests
Finished 5 requests
Finished 6 requests
Finished 7 requests
Finished 8 requests
Finished 9 requests
Finished 10 requests
Cost time 0.018644094467163086

这次比上次更快了，为什么呢？因为这次所有的请求都命中了 Cache，所以很快返回了结果。

后端配置

刚才我们知道了，requests-cache 默认使用了 SQLite 作为缓存对象，那这个能不能换啊？比如用文件，或者其他的数据库呢？

自然是可以的。

比如我们可以把后端换成本地文件，那可以这么做：

1
2
3
4
5
6
7
8
9
10
11
12
13

import time
import requests
import requests_cache

requests_cache.install_cache('demo_cache', backend='filesystem')

start = time.time()
session = requests.Session()
for i in range(10):
    session.get('http://httpbin.org/delay/1')
    print(f'Finished {i + 1} requests')
end = time.time()
print('Cost time', end - start)

这里我们添加了一个 backend 参数，然后指定为 filesystem，这样运行之后本地就会生成一个 demo_cache 的文件夹用作缓存，如果不想用缓存的话把这个文件夹删了就好了。

当然我们还可以更改缓存文件夹的位置，比如：

1

requests_cache.install_cache('demo_cache', backend='filesystem', use_temp=True)

这里添加一个 use_temp 参数，缓存文件夹便会使用系统的临时目录，而不会在代码区创建缓存文件夹。

当然也可以这样：

1

requests_cache.install_cache('demo_cache', backend='filesystem', use_cache_dir=True)

这里添加一个 use_cache_dir 参数，缓存文件夹便会使用系统的专用缓存文件夹，而不会在代码区创建缓存文件夹。

另外除了文件系统，requests-cache 也支持其他的后端，比如 Redis、MongoDB、GridFS 甚至内存，但也需要对应的依赖库支持，具体可以参见下表：

比如使用 Redis 就可以改写如下：

1
2

backend = requests_cache.RedisCache(host='localhost', port=6379)
requests_cache.install_cache('demo_cache', backend=backend)

更多详细配置可以参考官方文档：https://requests-cache.readthedocs.io/en/stable/user_guide/backends.html#backends。

Filter

当然，我们有时候也想指定有些请求不缓存，比如只缓存 POST 请求，不缓存 GET 请求，那可以这样来配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import time
import requests
import requests_cache

requests_cache.install_cache('demo_cache2', allowable_methods=['POST'])

start = time.time()
session = requests.Session()
for i in range(10):
    session.get('http://httpbin.org/delay/1')
    print(f'Finished {i + 1} requests')
end = time.time()
print('Cost time for get', end - start)
start = time.time()

for i in range(10):
    session.post('http://httpbin.org/delay/1')
    print(f'Finished {i + 1} requests')
end = time.time()
print('Cost time for post', end - start)

这里我们添加了一个 allowable_methods 指定了一个过滤器，只有 POST 请求会被缓存，GET 请求就不会。

看下运行结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

Finished 1 requests
Finished 2 requests
Finished 3 requests
Finished 4 requests
Finished 5 requests
Finished 6 requests
Finished 7 requests
Finished 8 requests
Finished 9 requests
Finished 10 requests
Cost time for get 12.916549682617188
Finished 1 requests
Finished 2 requests
Finished 3 requests
Finished 4 requests
Finished 5 requests
Finished 6 requests
Finished 7 requests
Finished 8 requests
Finished 9 requests
Finished 10 requests
Cost time for post 1.2473630905151367

这时候就看到 GET 请求由于没有缓存，就花了 12 多秒才结束，而 POST 由于使用了缓存，一秒多就结束了。

另外我们还可以针对 Response Status Code 进行过滤，比如只有 200 会缓存，则可以这样写：

1
2
3
4
5

import time
import requests
import requests_cache

requests_cache.install_cache('demo_cache2', allowable_codes=(200,))

当然我们还可以匹配 URL，比如针对哪种 Pattern 的 URL 缓存多久，则可以这样写：

1
2
3

urls_expire_after = {'*.site_1.com': 30, 'site_2.com/static': -1}
requests_cache.install_cache(
    'demo_cache2', urls_expire_after=urls_expire_after)

这样的话，site_1.com 的内容就会缓存 30 秒，site_2.com/static 的内容就永远不会过期。

当然，我们也可以自定义 Filter，具体可以参见：https://requests-cache.readthedocs.io/en/stable/user_guide/filtering.html#custom-cache-filtering。

Cache Headers

除了我们自定义缓存，requests-cache 还支持解析 HTTP Request / Response Headers 并根据 Headers 的内容来缓存。

比如说，我们知道 HTTP 里面有个 Cache-Control 的 Request / Response Header，它可以指定浏览器要不要对本次请求进行缓存，那 requests-cache 怎么来支持呢？

实例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

import time
import requests
import requests_cache

requests_cache.install_cache('demo_cache3')

start = time.time()
session = requests.Session()
for i in range(10):
    session.get('http://httpbin.org/delay/1',
                headers={
                    'Cache-Control': 'no-store'
                })
    print(f'Finished {i + 1} requests')
end = time.time()
print('Cost time for get', end - start)
start = time.time()

这里我们在 Request Headers 里面加上了 Cache-Control 为 no-store，这样的话，即使我们声明了缓存那也不会生效。

当然 Response Headers 的解析也是支持的，我们可以这样开启：

1

requests_cache.install_cache('demo_cache3', cache_control=True)

如果我们配置了这个参数，那么 expire_after 的配置就会被覆盖而不会生效。

更多的用法可以参见：https://requests-cache.readthedocs.io/en/stable/user_guide/headers.html#cache-headers。

总结

好了，到现在为止，一些基本配置、过期时间配置、后端配置、过滤器配置等基本常见的用法就介绍到这里啦，更多详细的用法大家可以参考官方文档：https://requests-cache.readthedocs.io/en/stable/user_guide.html。

希望对大家有帮助。

更多精彩内容，请关注我的公众号「进击的 Coder」和「崔庆才丨静觅」。

利用好搜索引擎

互联网时代，我们面临的是知识爆炸而不是知识匮乏。网上有很多很多好的学习资源，比如一些学习文档、疑难问题解决方案，很多都可以在网上搜到。

虽然网上有这些内容，但不同的搜索方法和用不同的搜索引擎搜到的结果就大不一样。

比如说，我们平时遇到了一些编程相关的问题，在谷歌中用英文搜索的结果在绝大多数情况下都会比在百度用中文搜索的结果好。比如说前者的结果通常就会是一些官方文档说明，而后者大多都是一些中文版 CSDN 博客，谁更前沿、更权威高下立判了。

我是做技术相关的，所以对于一些技术内容，我个人是非常建议首选谷歌英文搜索的，多数情况下能够更快更好地解决问题。

多看一手资料

我们知道，现在网上很多框架、工具，其都会配一个官方文档，比如 Python 的某个工具库、Vue 的的某个脚手架等等，同时很多源码也会在 GitHub 上公开。

我们如果要进行学习这些内容的话，我个人推荐尽量多去查询一些一手的资料，比如一些入门使用方法，可以尽量去看官方文档的一些 Get Started 部分；比如一些疑难 Issues，可以去 GitHub Issues 区搜索下关键词。

当然有的同学会说，官方文档都是些英文的，我看不懂啊，所以通常都会去搜索一些博客文章来看，比如一些中文博客的教程。可以是可以，也的确有一些优秀的博主能写出一些不错的文章，但毕竟还是少，而且这永远都不是一手资料，多数情况下，博客的文章也会有一些实效性的问题或者难免会出现一些错误。

所以，我个人还是推荐尽量去看一手资料。但一手的资料通常英文居多，但还是建议大家能够尽量地适应去读英文文档，如果能够做到的话，我们获取知识的能力会继续上一个台阶。

时间管理和分类

每个人的精力都是有限的，一些工作和其他的琐事可以说是无穷无尽的。

所以，在做一些工作和学习的事情的时候，我们需要去区分优先级和重要程度，也就是能够合理地管理好自己要做的事情。

我个人会用一个清单软件（我用的滴答清单）来记录我所有需要做的事情，然后会给每个事情进行分类，比如说我会划分工作、学习、私人、购物、电影等等各种分类，然后每个任务都会指定好优先级和过期事件。指定好了之后，清单软件有一些功能可以给我筛选出来哪些是紧急重要，哪些是不紧急重要，哪些是紧急不重要等等的事情，然后我会有选择地去做对应的事情。比如说我会把大量的时间花在重要的事情上，不紧急不重要的可以看看能否尽量规避或者找人代做，总之不同的类型需要有不同的应对方案。

另外还有一些习惯养成类的事情，比如说定期的学习计划、定期的健身、定期的冥想、定期的跑步，也可以列入到个人的事件规划中。我通常以打卡的形式记录在清单软件中，每天都会有定期提醒，这样做完之后我就会打一次卡，看着越来越满的打卡记录，会感觉比较有成就感，大家也可以来试试。

要有一个短期目标

我们有时候做事的时候，脑子里知道很多长远的目标是什么。比如说，我长远计划里面有一个事情是要做一个网站系统，这是一个大目标，同时也有一个长远计划是要学习精通一门编程语言，这也是一个大目标。很多大目标都在我们的潜意识里面存留着。

现在问一个问题，虽然这些大的目标都在我们脑海里，但有没有一个时间，自己突然闲下来或者临时没有事的时候，却不知道这个空闲的时间去做什么？

如果有，那很可能就是因为没有短期目标。因为这个目标在我们的脑海中太大了，根本无法落实到执行的地步，所以我们需要做的事就是把一些目标进行拆解，拆解到什么地步呢？拆解到能够想到就能立马开始做的地步，这就是一些短期目标。

比如说，我们要学习一门课，我们可以给自己列个计划，比如哪天可以看哪个视频，或者一篇文章，这是知道了就能立马去做的事情。

所以，有了这个短期目标，我们能够更好地落实到执行上，这也是能够有效延缓拖延的方法。

不要完美主义

在做一些事情的时候，我们不要过分地追求完美主义。不是说不好，是因为这样很容易消磨我们的精力和耐心。

比如说，我们学习背单词吧，比如每天的计划是 20 个单词，好第一天背了 20 个，然后接着第二天的时候发现前 20 个单词没有背过，然后就接着背前 20 个单词，然后第三天的时候发现第一天和第二天的 40 个还是记得不牢固，然后就觉得好难，最后就放弃了，这就是因为过分追求完美主义导致的问题。

学习并不是非 0 即 1 的，我们如果能够学会 20%、60%、80% 也是一个不错的进步。

所以，我们不要执着于完美主义，非要做到 100% 不可，这样会把自己的精力和耐心慢慢消磨，直到放弃。

不是所有教材都适合每个人

并不是所有权威教材都是适合每个人的，要去寻找适合自己的学习方式。

市面上其实有很多所谓的权威教材或者网红教材，但这些教材并不是万能的，众口是很难调的，因为每个人的基础、水平都是不同的。

比如说一本书里面在前面的章节写了一些基础的环境配置和基础知识，有些人就会觉得非常友好，会觉得非常实用，但有些人就会觉得非常啰嗦，没有重点。比如说有人在学习一个框架和库的时候就喜欢看视频学习，因为这样能够看到具体的操作流程，但有些人就会觉得看视频学习非常浪费时间，而且知识点不好找，还是看官方文档或看书更方便。

这些学习方法和偏好没有绝对的对与不对，我们也不用非要跟风去购买和学习某个特定的教材和学习形式，适合自己的才是最好的。

多进行总结和记录

这个是非常非常非常重要的，在学习的过程中把学习笔记记录下来是一个非常好而且有效的学习习惯。

好处有这么以下几点：

• 自己的学习笔记是对自己学习过程的梳理和总结，梳理和总结的过程就是一个学习复盘的过程，能够加深自己对知识点的印象。
• 方便复习会看，好记性不如烂笔头。写下来之后，如果我们想要对某个知识点进行复习，是非常容易的，因为文章的整体思路本身就是自己的，要捡起来也非常容易。
• 如果我们能够把学习内容整理发表出去，大家也可以对文章进行阅读和评论，在讨论的过程中可以有更多思维火花的碰撞，说不定能有更深入的了解。
• 能够帮助更多的人，因为我们遇到的问题通常也是别人遇到的，如果能够帮助更多的人，心里肯定也是很有成就感的。

学习要有深有浅

学习一个知识点，我们也是需要有深浅的控制的，也是需要评估一些学习时间和成本的。

比如一个知识点，我们可以给它划分成三个层级，第一层级是会用即可，第二层级是熟练运用，第三层是深刻理解并改写。

在我们日常的工作中，由于不同技术栈和项目的需要，对一些知识的需求也会不一样，比如一些核心的技术，我们就需要深入理解并改写。比如说假如我是做 Scrapy 爬虫的，那对于 Scrapy 框架我就需要做到第三个层次，即深入理解并能改写；对于一些较高频的工具，比如 argparse，那我们就需要做到熟练运用；但对于一些低频且比较边角的知识点，我们只要花最少的时间知道它最基本的用法就好了，因为可能我们就是用到了它的最基本的用法解决了一个边角问题，所以没必要花太多时间在上面。

所以，对于一些学习内容，我们要能够分清楚这个知识点应该学到什么地步，然后采取对应的学习方案。

路径依赖

我们在学习的时候要尽量避免一些路径依赖的问题。

比如说，一位同学要学习 Python 机器学习相关内容，Python 机器学习的基础是一些 Python 和数学相关的内容，那他就非要把 Python 和数学的知识先全部研究透，比如说把所有的 Python 基础全学一遍、把所有的高等数学、统计学的知识全都学一遍，然后再回过头来学习 Python 机器学习，结果学习的时候发现很多 Python 基础和数学基础都用不到，然后久而久之，用不到的 Python 基础和数学基础就慢慢忘记了，而且 Python 机器学习的学习周期也被大大拉长。

这个例子中出现的就是路径依赖问题，我们其实没必要非要把所有的依赖项都完美一个个地彻底解决了再来学习对应的知识，知识点都是有关联的，我们在学习的时候可以以最终的结果为导向。

比如说，我今天要学 Python 机器学习，比如一个分类算法的实现，那我就把 Python 的模型定义、类定义、方法定义学会，同时研究好数学中的分类算法的思路，那就可以去学习 Python 机器学习了，这样整体效果也会更快更好，同时学习到的知识也能够用得上，且紧密关联。

学习优秀的源码

很多很多优秀的编程思路和方法其实都隐藏在一些优秀的源代码库里面。

比如说，学习爬虫，Scrapy 框架为什么能够做到这么好的扩展性？比如说，学习网站开发，Vue 为什么能够吸引这么多开发者学习？这其中都是有一定原因的，这些优秀的框架也是有它们的过人之处的，另外一些优秀的源码里面通常质量也会很高。

所以，我们如果能够多去阅读一些优秀的框架或库的源码，能够学到很多有用的编程思路和技巧的，如果能够把这些思路和知识运用到自己的工作和项目中，那一定会大有帮助。

实践很重要

这个就不用多说了，光说不练，等于白搭。

对于我们做技术的来说，如果我们只是干巴巴地阅读一些官方文档和教程，而不去实际编写一些代码运行的话，收获是很少的。

一般来说，如果我们学习一些框架和库的时候，如果能够跟着把一些样例敲下来，真的能够理解深入很多。通常，阅读的时候我们不会发现问题，但一但一点点跟着敲下来，把代码运行起来，我们会发现很多潜在的问题，而且会对问题的认识更加深刻。

还有就是，遇到问题的时候，我们也需要多去实践和探索，如果不是十分紧急，我们可以尽量去尝试去搜索问题的解决方案，去 debug，去找 root cause，这样我们就能对某个问题有更加深刻的认识，同时自己解决问题的能力也会大大提高。

贵有恒

是的，做一件事或者学习一个知识，一个非常非常重要的要素就是有恒心，即坚持。

贵有恒，何必三更起五更睡。

是的，做成一件事一个很大的拦路虎就是半途而废、三天打鱼两天晒网，这样很容易做着做着就没有下文了，然后就再也没有然后了，很多很多的事情就是因为这个而失败了。

贵有恒，坚持下来，做好计划，一件事，如果我们能够坚持做下来，一天天慢慢积累，其威力是无穷的。

更多精彩内容，请关注我的公众号「进击的 Coder」和「崔庆才丨静觅」。

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3 网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

在上一节我们了解了网站登录验证和模拟登录的基本原理。网站登录验证主要有两种实现方式，一种是基于 Session + Cookies 的登录验证，另一种是基于 JWT 的登录验证。接下来两节，我们就通过两个实例来分别讲解这两种登录验证的分析和模拟登录流程。

本节主要介绍 Session + Cookie 模拟登录的流程。

1. 准备工作

在本节开始之前，我们需要先做好如下准备工作。

• 安装好了 requests 请求库并学会了其基本用法。
• 安装好了 Selenium 库并学会了其基本用法。

下面我们就用两个案例来分别讲解模拟登录的实现。

2. 案例介绍

本节有一个适用于 Session + Cookie 模拟登录的案例网站，网址为：https://login2.scrape.center/，访问之后，我们会看到一个登录页面，如图所示：

我们输入用户名和密码（用户名和密码都是 admin），然后点击登录。登录成功后，我们便可以看到一个和之前案例类似的电影网站，如图所示。

这个网站是基于传统的 MVC 模式开发的，因此也比较适合 Session + Cookie 的模拟登录。

3. 模拟登录

对于这个网站，我们如果要模拟登录，就需要先分析登录过程究竟发生了什么。我们打开开发者工具，重新执行登录操作，查看其登录过程中发生的请求，如图所示。

图 10-5 登录过程中发生的请求

从图 10-5 中我们可以看到，在登录的瞬间，浏览器发起了一个 POST 请求，目标 URL 为 https://login2.scrape.center/login，并通过表单提交的方式像服务器提交了登录数据，其中包括 username 和 password 两个字段，返回的状态码是 302，Response Headers 的 location 字段为根页面，同时 Response Headers 还包含了 set-cookie 信息，设置了 Session ID。

由此我们可以发现，要实现模拟登录，我们只需要模拟这个请求就好了。登录完成后获取 Response 设置的 Cookie，将它保存好，后续发出请求的时候带上 Cookies 就可以正常访问了。

好，那么我们就来用代码实现一下吧！

在默认情况下，每次 requests 请求都是独立且互不干扰的，比如我们第一次调用了 post 方法模拟登录了一下，紧接着再调用 get 方法请求主页面。其实这是两个完全独立的请求，第一次请求获取的 Cookie 并不能传给第二次请求，因此常规的顺序调用是不能起到模拟登录效果的。

我们来看一段无效的代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

import requests
from urllib.parse import urljoin

BASE_URL = 'https://login2.scrape.center/'
LOGIN_URL = urljoin(BASE_URL, '/login')
INDEX_URL = urljoin(BASE_URL, '/page/1')
USERNAME = 'admin'
PASSWORD = 'admin'

response_login = requests.post(LOGIN_URL, data={
    'username': USERNAME,
    'password': PASSWORD
})

response_index = requests.get(INDEX_URL)
print('Response Status', response_index.status_code)
print('Response URL', response_index.url)

这里我们先定义了几个基本的 URL 、用户名和密码，然后我们分别用 requests 请求了登录的 URL 进行模拟登录，紧接着请求了首页来获取页面内容，能正常获取数据吗？由于 requests 可以自动处理重定向，我们可以在最后把 Response 的 URL 打印出来，如果它的结果是 INDEX_URL，那么证明模拟登录成功并成功爬取到了首页的内容。如果它跳回到了登录页面，那就说明模拟登录失败。

我们通过结果来验证一下，运行结果如下：

1
2

Response Status 200
Response URL https://login2.scrape.center/login?next=/page/1

这里可以看到，其最终的页面 URL 是登录页面的 URL。另外这里也可以通过 Response 的 text 属性来验证下页面源码，其源码内容就是登录页面的源码内容，由于内容较多，这里就不再输出比对了。

总之，这个现象说明我们并没有成功完成模拟登录，这是因为 requests 直接调用 post、get 等方法，每次请求都是一个独立的请求，都相当于是新开了一个浏览器打开这些链接，所以这两次请求对应的 Session 并不是同一个，这里我们模拟了第一个 Session 登录，并不能影响第二个 Session 的状态，因此模拟登录也就无效了。

那么怎样才能实现正确的模拟登录呢？

我们知道 Cookie 里面是保存了 Session ID 信息的，刚才也观察到了登录成功后 Response Headers 里面有 set-cookie 字段，实际上这就是让浏览器生成了 Cookie。因为 Cookies 里面包含了 Session ID 的信息，所以只要后续的请求带着这些 Cookie，服务器便能通过 Cookie 里的 Session ID 信息找到对应的 Session 了，因此，服务端对于这两次请求就会使用同一个 Session 了。因为第一次我们已经成功完成了模拟登录，所以 Session 里面就记录了用户的登录信息，在第二次访问的时候，由于是同一个 Session，服务器就能知道用户当前是登录状态，那就能够返回正确的结果而不再是跳转到登录页面了。

所以，这里的关键在于两次请求的 Cookie 的传递。这里我们可以把第一次模拟登录后的 Cookie 保存下来，在第二次请求的时候加上这个 Cookie，代码可以改写如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

import requests
from urllib.parse import urljoin

BASE_URL = 'https://login2.scrape.center/'
LOGIN_URL = urljoin(BASE_URL, '/login')
INDEX_URL = urljoin(BASE_URL, '/page/1')
USERNAME = 'admin'
PASSWORD = 'admin'

response_login = requests.post(LOGIN_URL, data={
    'username': USERNAME,
    'password': PASSWORD
}, allow_redirects=False)

cookies = response_login.cookies
print('Cookies', cookies)

response_index = requests.get(INDEX_URL, cookies=cookies)
print('Response Status', response_index.status_code)
print('Response URL', response_index.url)

由于 requests 可以自动处理重定向，所以我们模拟登录的过程要加上 allow_redirects 参数并将其设置为 False，使其不自动处理重定向。我们将登录之后返回的 Response 赋值为 response_login，这样调用 response_login 的 cookies 就是获取了网站的 Cookie 信息了。这里 requests 自动帮我们解析了 Response Headers 的 set-cookie 字段并设置了 Cookie，所以我们不用再去手动解析 Response Headers 的内容了，直接使用 response_login 对象的 cookies 方法即可获取 Cookie。

好，接下来我们再次用 requests 的 get 方法来请求网站的 INDEX_URL。不过这里和之前不同，get 方法增加了一个参数 cookies，这就是第一次模拟登录完之后获取的 Cookie，这样第二次请求就能携带第一次模拟登录获取的 Cookie 信息了，此时网站会根据 Cookie 里面的 Session ID 信息查找到同一个 Session，校验其已经是登录状态，然后返回正确的结果。

这里我们还是输出最终的 URL，如果它是 INDEX_URL，就代表模拟登录成功并获取了有效数据，否则就代表模拟登录失败。

我们看下运行结果：

1
2
3

Cookies <RequestsCookieJar[<Cookie sessionid=psnu8ij69f0ltecd5wasccyzc6ud41tc for login2.scrape.center/>]>
Response Status 200
Response URL https://login2.scrape.center/page/1

这下没有问题了，我们发现其 URL 就是 INDEX_URL，模拟登录成功了！同时还可以进一步输出 response_index 的 text 属性看下是否获取成功。

后续用同样的方式爬取即可。但其实我们发现，这种实现方式比较烦琐，每次还需要处理 Cookie 并一次传递，有没有更简便的方法呢？

有的，我们可以直接借助于 requests 内置的 Session 对象来帮我们自动处理 Cookie，使用了 Session 对象之后，requests 会自动保存每次请求后需要设置的 Cookie ，并在下次请求时自动携带它，就相当于帮我们维持了一个 Session 对象，这样就更方便了。

所以，刚才的代码可以简化如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

import requests
from urllib.parse import urljoin

BASE_URL = 'https://login2.scrape.center/'
LOGIN_URL = urljoin(BASE_URL, '/login')
INDEX_URL = urljoin(BASE_URL, '/page/1')
USERNAME = 'admin'
PASSWORD = 'admin'

session = requests.Session()

response_login = session.post(LOGIN_URL, data={
    'username': USERNAME,
    'password': PASSWORD
})

cookies = session.cookies
print('Cookies', cookies)

response_index = session.get(INDEX_URL)
print('Response Status', response_index.status_code)
print('Response URL', response_index.url)

可以看到，这里我们无须再关心 Cookie 的处理和传递问题，我们声明了一个 Session 对象，然后每次调用请求的时候都直接使用 Session 对象的 post 或 get 方法就好了。

运行效果是完全一样的，结果如下：

1
2
3

Cookies <RequestsCookieJar[<Cookie sessionid=ssngkl4i7en9vm73bb36hxif05k10k13 for login2.scrape.center/>]>
Response Status 200
Response URL https://login2.scrape.center/page/1

因此，为了简化写法，这里建议直接使用 Session 对象进行请求，这样我们无须关心 Cookie 的操作了，实现起来会更加方便。

这个案例整体来说比较简单，但是如果碰上复杂一点的网站，如带有验证码，带有加密参数等，直接用 requests 并不好处理模拟登录，如果登录不了，那整个页面不就都没法爬取了吗？有没有其他的方式来解决这个问题呢？当然是有的，比如说我们可以使用 Selenium 来模拟浏览器，进而实现模拟登录，然后获取模拟登录成功后的 Cookie，再把获取的 Cookie 交由 requests 等来爬取就好了。

这里我们还是以刚才的页面为例，把模拟登录这块交由 Selenium 来实现，后续的爬取交由 requests 来实现，相关的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

from urllib.parse import urljoin
from selenium import webdriver
import requests
import time

BASE_URL = 'https://login2.scrape.center/'
LOGIN_URL = urljoin(BASE_URL, '/login')
INDEX_URL = urljoin(BASE_URL, '/page/1')
USERNAME = 'admin'
PASSWORD = 'admin'

browser = webdriver.Chrome()
browser.get(BASE_URL)
browser.find_element_by_css_selector('input[name="username"]').send_keys(USERNAME)
browser.find_element_by_css_selector('input[name="password"]').send_keys(PASSWORD)
browser.find_element_by_css_selector('input[type="submit"]').click()
time.sleep(10)

# get cookies from selenium
cookies = browser.get_cookies()
print('Cookies', cookies)
browser.close()

# set cookies to requests
session = requests.Session()
for cookie in cookies:
    session.cookies.set(cookie['name'], cookie['value'])

response_index = session.get(INDEX_URL)
print('Response Status', response_index.status_code)
print('Response URL', response_index.url)

这里我们使用 Selenium 先打开了 Chrome，然后跳转到了登录页面，随后模拟输入了用户名和密码，接着点击了登录按钮，我们可以发现浏览器提示登录成功，然后跳转到了主页面。

这时候，我们通过调用 get_cookies 方法便能获取当前浏览器所有的 Cookie，这就是模拟登录成功之后的 Cookie，用这些 Cookie 我们就能访问其他数据了。

接下来，我们声明了 requests 的 Session 对象，然后遍历了刚才的 Cookie 并将其设置到 Session 对象的 cookies 属性上，接着再拿着这个 Session 对象去请求 INDEX_URL，就也能够获取对应的信息而不会跳转到登录页面了。

运行结果如下：

1
2
3

Cookies [{'domain': 'login2.scrape.center', 'expiry': 1589043753.553155, 'httpOnly': True, 'name': 'sessionid', 'path': '/', 'sameSite': 'Lax', 'secure': False, 'value': 'rdag7ttjqhvazavpxjz31y0tmze81zur'}]
Response Status 200
Response URL https://login2.scrape.center/page/1

可以看到，这里的模拟登录和后续的爬取也成功了。所以说，如果碰到难以模拟登录的过程，我们也可以使用 Selenium 等模拟浏览器的操作方式来实现，其目的就是获取登录后的 Cookie，有了 Cookie 之后，我们再用这些 Cookie 爬取其他页面就好了。

所以这里我们也可以发现，对于基于 Session + Cookie 验证的网站，模拟登录的核心要点就是获取 Cookie。这个 Cookie 可以被保存下来或传递给其他的程序继续使用，甚至可以将 Cookie 持久化存储或传输给其他终端来使用。

另外，为了提高 Cookie 利用率或降低封号概率，可以搭建一个账号池实现 Cookie 的随机取用。

4. 总结

以上我们通过一个示例来演示了模拟登录爬取的过程，以后遇到这种情形的时候就可以用类似的思路解决了。

本节代码：https://github.com/Python3WebSpider/ScrapeLogin2。

系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

在上一节中，我们介绍了异步爬虫的基本原理和 asyncio 的基本用法，并且在最后简单提及了使用 aiohttp 来实现网页爬取的过程。在本节中，我们来介绍一下 aiohttp 的常见用法。

1. 基本介绍

前面介绍的 asyncio 模块内部实现了对 TCP、UDP、SSL 协议的异步操作，但是对于 HTTP 请求来说，我们就需要用到 aiohttp 来实现了。

aiohttp 是一个基于 asyncio 的异步 HTTP 网络模块，它既提供了服务端，又提供了客户端。其中我们用服务端可以搭建一个支持异步处理的服务器，就是用来处理请求并返回响应的，类似于 Django、Flask、Tornado 等一些 Web 服务器。而客户端可以用来发起请求，类似于使用 requests 发起一个 HTTP 请求然后获得响应，但 requests 发起的是同步的网络请求，aiohttp 则是异步的。

本节中，我们主要了解一下 aiohttp 客户端部分的用法。

2. 基本实例

首先，我们来看一个基本的 aiohttp 请求案例，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text(), response.status

async def main():
    async with aiohttp.ClientSession() as session:
        html, status = await fetch(session, 'https://cuiqingcai.com')
        print(f'html: {html[:100]}...')
        print(f'status: {status}')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

这里我们使用 aiohttp 来爬取我的个人博客，获得了源码和响应状态码并输出出来，运行结果如下：

1
2
3
4
5
6

html: <!DOCTYPE HTML>
<html>
<head>
<meta charset="UTF-8">
<meta name="baidu-tc-verification" content=...
status: 200

这里网页源码过长，只截取输出了一部分。可以看到，这里我们成功获取了网页的源代码及响应状态码 200，也就完成了一次基本的 HTTP 请求，即我们成功使用 aiohttp 通过异步的方式来进行了网页爬取。当然，这个操作用之前讲的 requests 也可以做到。

可以看到，其请求方法的定义和之前有了明显的区别，主要有如下几点：

• 首先在导入库的时候，我们除了必须要引入 aiohttp 这个库之外，还必须要引入 asyncio 这个库。因为要实现异步爬取，需要启动协程，而协程则需要借助于 asyncio 里面的事件循环来执行。除了事件循环，asyncio 里面也提供了很多基础的异步操作。
• 异步爬取方法的定义和之前有所不同，在每个异步方法前面统一要加 async 来修饰。
• with as 语句前面同样需要加 async 来修饰。在 Python 中，with as 语句用于声明一个上下文管理器，能够帮我们自动分配和释放资源。而在异步方法中，with as 前面加上 async 代表声明一个支持异步的上下文管理器。
• 对于一些返回 coroutine 的操作，前面需要加 await 来修饰。比如 response 调用 text 方法，查询 API 可以发现，其返回的是 coroutine 对象，那么前面就要加 await；而对于状态码来说，其返回值就是一个数值类型，那么前面就不需要加 await。所以，这里可以按照实际情况处理，参考官方文档说明，看看其对应的返回值是怎样的类型，然后决定加不加 await 就可以了。
• 最后，定义完爬取方法之后，实际上是 main 方法调用了 fetch 方法。要运行的话，必须要启用事件循环，而事件循环就需要使用 asyncio 库，然后使用 run_until_complete 方法来运行。

注意：在 Python 3.7 及以后的版本中，我们可以使用 asyncio.run(main()) 来代替最后的启动操作，不需要显示声明事件循环，run 方法内部会自动启动一个事件循环。但这里为了兼容更多的 Python 版本，依然还是显式声明了事件循环。

3. URL 参数设置

对于 URL 参数的设置，我们可以借助于 params 参数，传入一个字典即可，示例如下：

1
2
3
4
5
6
7
8
9
10
11

import aiohttp
import asyncio

async def main():
    params = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/get', params=params) as response:
            print(await response.text())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

{
  "args": {
    "age": "25",
    "name": "germey"
  },
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Host": "httpbin.org",
    "User-Agent": "Python/3.7 aiohttp/3.6.2",
    "X-Amzn-Trace-Id": "Root=1-5e85eed2-d240ac90f4dddf40b4723ef0"
  },
  "origin": "17.20.255.122",
  "url": "https://httpbin.org/get?name=germey&age=25"
}

这里可以看到，其实际请求的 URL 为 https://httpbin.org/get?name=germey&age=25，其 URL 请求参数就对应了 params 的内容。

4. 其他请求类型

另外，aiohttp 还支持其他请求类型，如 POST、PUT、DELETE 等，这和 requests 的使用方式有点类似，示例如下：

1
2
3
4
5
6

session.post('http://httpbin.org/post', data=b'data')
session.put('http://httpbin.org/put', data=b'data')
session.delete('http://httpbin.org/delete')
session.head('http://httpbin.org/get')
session.options('http://httpbin.org/get')
session.patch('http://httpbin.org/patch', data=b'data')

要使用这些方法，只需要把对应的方法和参数替换一下即可。

5. POST 请求

对于 POST 表单提交，其对应的请求头的 Content-Type 为 application/x-www-form-urlencoded，我们可以用如下方式来实现，代码示例如下：

1
2
3
4
5
6
7
8
9
10
11

import aiohttp
import asyncio

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://httpbin.org/post', data=data) as response:
            print(await response.text())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

{
  "args": {},
  "data": "",
  "files": {},
  "form": {
    "age": "25",
    "name": "germey"
  },
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Content-Length": "18",
    "Content-Type": "application/x-www-form-urlencoded",
    "Host": "httpbin.org",
    "User-Agent": "Python/3.7 aiohttp/3.6.2",
    "X-Amzn-Trace-Id": "Root=1-5e85f0b2-9017ea603a68dc285e0552d0"
  },
  "json": null,
  "origin": "17.20.255.58",
  "url": "https://httpbin.org/post"
}

对于 POST JSON 数据提交，其对应的请求头的 Content-Type 为 application/json，我们只需要将 post 方法的 data 参数改成 json 即可，代码示例如下：

1
2
3
4
5

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://httpbin.org/post', json=data) as response:
            print(await response.text())

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

{
  "args": {},
  "data": "{\"name\": \"germey\", \"age\": 25}",
  "files": {},
  "form": {},
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Content-Length": "29",
    "Content-Type": "application/json",
    "Host": "httpbin.org",
    "User-Agent": "Python/3.7 aiohttp/3.6.2",
    "X-Amzn-Trace-Id": "Root=1-5e85f03e-c91c9a20c79b9780dbed7540"
  },
  "json": {
    "age": 25,
    "name": "germey"
  },
  "origin": "17.20.255.58",
  "url": "https://httpbin.org/post"
}

可以发现，其实现也和 requests 非常像，不同的参数支持不同类型的请求内容。

6. 响应

对于响应来说，我们可以用如下方法分别获取响应的状态码、响应头、响应体、响应体二进制内容、响应体 JSON 结果，示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

import aiohttp
import asyncio

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://httpbin.org/post', data=data) as response:
            print('status:', response.status)
            print('headers:', response.headers)
            print('body:', await response.text())
            print('bytes:', await response.read())
            print('json:', await response.json())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

status: 200
headers: <CIMultiDictProxy('Date': 'Thu, 02 Apr 2020 14:13:05 GMT', 'Content-Type': 'application/json', 'Content-Length': '503', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true')>
body: {
  "args": {},
  "data": "",
  "files": {},
  "form": {
    "age": "25",
    "name": "germey"
  },
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Content-Length": "18",
    "Content-Type": "application/x-www-form-urlencoded",
    "Host": "httpbin.org",
    "User-Agent": "Python/3.7 aiohttp/3.6.2",
    "X-Amzn-Trace-Id": "Root=1-5e85f2f1-f55326ff5800b15886c8e029"
  },
  "json": null,
  "origin": "17.20.255.58",
  "url": "https://httpbin.org/post"
}

bytes: b'{\n  "args": {}, \n  "data": "", \n  "files": {}, \n  "form": {\n    "age": "25", \n    "name": "germey"\n  }, \n  "headers": {\n    "Accept": "*/*", \n    "Accept-Encoding": "gzip, deflate", \n    "Content-Length": "18", \n    "Content-Type": "application/x-www-form-urlencoded", \n    "Host": "httpbin.org", \n    "User-Agent": "Python/3.7 aiohttp/3.6.2", \n    "X-Amzn-Trace-Id": "Root=1-5e85f2f1-f55326ff5800b15886c8e029"\n  }, \n  "json": null, \n  "origin": "17.20.255.58", \n  "url": "https://httpbin.org/post"\n}\n'
json: {'args': {}, 'data': '', 'files': {}, 'form': {'age': '25', 'name': 'germey'}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Content-Length': '18', 'Content-Type': 'application/x-www-form-urlencoded', 'Host': 'httpbin.org', 'User-Agent': 'Python/3.7 aiohttp/3.6.2', 'X-Amzn-Trace-Id': 'Root=1-5e85f2f1-f55326ff5800b15886c8e029'}, 'json': None, 'origin': '17.20.255.58', 'url': 'https://httpbin.org/post'}

这里我们可以看到有些字段前面需要加 await，有的则不需要。其原则是，如果它返回的是一个 coroutine 对象（如 async 修饰的方法），那么前面就要加 await，具体可以看 aiohttp 的 API，其链接为：https://docs.aiohttp.org/en/stable/client_reference.html。

7. 超时设置

对于超时设置，我们可以借助 ClientTimeout 对象，比如这里要设置 1 秒的超时，可以这么实现：

1
2
3
4
5
6
7
8
9
10
11

import aiohttp
import asyncio

async def main():
    timeout = aiohttp.ClientTimeout(total=1)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get('https://httpbin.org/get') as response:
            print('status:', response.status)

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

如果在 1 秒之内成功获取响应的话，运行结果如下：

1

200

如果超时的话，会抛出 TimeoutError 异常，其类型为 asyncio.TimeoutError，我们再进行异常捕获即可。

另外，声明 ClientTimeout 对象时还有其他参数，如 connect、socket_connect 等，详细可以参考官方文档：https://docs.aiohttp.org/en/stable/client_quickstart.html#timeouts。

8. 并发限制

由于 aiohttp 可以支持非常大的并发，比如上万、十万、百万都是能做到的，但对于这么大的并发量，目标网站很可能在短时间内无法响应，而且很可能瞬时间将目标网站爬挂掉，所以我们需要控制一下爬取的并发量。

一般情况下，我们可以借助于 asyncio 的 Semaphore 来控制并发量，示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import asyncio
import aiohttp

CONCURRENCY = 5
URL = 'https://www.baidu.com'

semaphore = asyncio.Semaphore(CONCURRENCY)
session = None

async def scrape_api():
    async with semaphore:
        print('scraping', URL)
        async with session.get(URL) as response:
            await asyncio.sleep(1)
            return await response.text()

async def main():
    global session
    session = aiohttp.ClientSession()
    scrape_index_tasks = [asyncio.ensure_future(scrape_api()) for _ in range(10000)]
    await asyncio.gather(*scrape_index_tasks)


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

这里我们声明了 CONCURRENCY（代表爬取的最大并发量）为 5，同时声明爬取的目标 URL 为百度。接着，我们借助于 Semaphore 创建了一个信号量对象，将其赋值为 semaphore，这样我们就可以用它来控制最大并发量了。怎么使用呢？这里我们把它直接放置在对应的爬取方法里面，使用 async with 语句将 semaphore 作为上下文对象即可。这样的话，信号量可以控制进入爬取的最大协程数量，即我们声明的 CONCURRENCY 的值。

在 main 方法里面，我们声明了 10000 个 task，将其传递给 gather 方法运行。倘若不加以限制，这 10000 个 task 会被同时执行，并发数量太大。但有了信号量的控制之后，同时运行的 task 的数量最大会被控制在 5 个，这样就能给 aiohttp 限制速度了。

9. 总结

本节我们了解了 aiohttp 的基本使用方法，更详细的内容还是推荐大家到官方文档查阅，详见 https://docs.aiohttp.org/。

本节代码：https://github.com/Python3WebSpider/AsyncTest。

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

很多情况下，一些网站的页面或资源我们通常需要登录才能看到。比如说访问 GitHub 的个人设置页面，如果不登录是无法查看的；比如说 12306 买票提交订单的页面，如果不登录是无法提交订单的；比如说要发一条微博，如果不登录是无法发送的。

我们之前学习的案例都是爬取的无需登录即可访问的站点，但是诸如上面例子的情况非常非常多，那假如我们想要用爬虫来访问这些页面，比如用爬虫修改 GitHub 的个人设置，用爬虫提交购票订单，用爬虫发微博，能做到吗？

答案是可以，这里就需要用到一些模拟登录相关的技术了。

那么本节我们就先来了解一下模拟登录的一些基本原理和实现吧。

1. 网站登录验证的实现

我们要实现模拟登录，那就得首先了解网站登录验证的实现。

登录一般是需要两个内容，用户名和密码，有的网站可能是手机号和验证码，有的是微信扫码，有的是 OAuth 验证等等，但根本上来说，都是把一些可供认证的信息提交给了服务器。

比如这里我们就拿用户名和密码来说吧。用户在一个网页表单里面输入了这些内容，然后点击登录按钮的一瞬间，浏览器客户端就会向服务器发送一个登录请求，这个请求里面肯定就包含了用户名和密码信息，这时候，服务器需要处理一下这些信息，然后返回给客户端一个类似「凭证」的东西，有了这个「凭证」以后呢，客户端拿着这个「凭证」再去访问某些需要登录才能查看的页面，服务器自然就能”放行“了，返回对应的内容或执行对应的操作就好了。

形象点说呢，我们拿登录发微博和买票坐火车这两件事来类比。发微博就好像要坐火车，没票是没法坐火车的吧，要坐火车怎么办呢？当然是先买票了，我们拿钱去火车站买个票，有了票之后，进站口查验一下，没问题就自然能去坐火车了，这个票就是坐火车的「凭证」。那发微博也一样，我们有用户名和密码，请求下服务器，获得一个「凭证」，这就相当于买到了火车票，然后在发微博的时候拿着这个「凭证」去请求服务器，服务器校验没问题，自然就把微博发出去了。

那么问题来了，这个「凭证」到底是怎么生成和验证的呢？目前比较流行的实现方式有两种，一种是基于 Session + Cookie 的验证，一种是基于 JWT（JSON Web Token）的验证，下面我们来介绍下。

2. Session 和 Cookie

我们在第一章了解了 Session 和 Cookie 的基本概念。简而言之呢，Session 就是存在服务端的，里面保存了用户此次访问的会话信息，Cookie 则是保存在用户本地浏览器的，它会在每次用户访问网站的时候发送给服务器，Cookie 会作为 Request Headers 的一部分发送给服务器，服务器根据 Cookie 里面包含的信息判断找出其 Session 对象并做一些校验，不同的 Session 对象里面维持了不同访问用户的状态，服务器可以根据这些信息决定返回 Response 的内容。

我们以用户登录的情形来说吧，其实不同的网站对于用户的登录状态的实现是可能不同的，但是 Session 和 Cookie 一定是相互配合工作的。

下面梳理如下：

• 比如说，Cookie 里面可能只存了 Session ID 相关信息，服务器能根据 Cookie 找到对应的 Session，用户登录之后，服务器会把对应的 Session 里面标记一个字段，代表已登录状态或者其他信息（如角色、登录时间）等等，这样用户每次访问网站的时候都带着 Cookie 来访问，服务器就能找到对应的 Session，然后看一下 Session 里面的状态是登录状态，那就可以返回对应的结果或执行某些操作。
• 当然 Cookie 里面也可能直接存了某些凭证信息。比如说用户在发起登录请求之后，服务器校验通过，返回给客户端的 Response Headers 里面可能带有 Set-Cookie 字段，里面可能就包含了类似凭证的信息，这样客户端会执行设置 Cookie 的操作，将这些信息保存到 Cookie 里面，以后再访问网页时携带这些 Cookie 信息，服务器拿着这里面的信息校验，自然也能实现登录状态检测了。

以上两种情况几乎能涵盖大部分的 Session 和 Cookie 登录验证的实现，具体的实现逻辑因服务器而异，但 Session 和 Cookie 一定是需要相互配合才能实现的。

3. JWT

Web 开发技术是一直在发展的，近几年前后端分离的趋势越来越火，很多 Web 网站都采取了前后端分离的技术来实现。而且传统的基于 Session 和 Cookie 的校验也存在一定问题，比如服务器需要维护登录用户的 Session 信息，而且分布式部署不方便，也不太适合前后端分离的项目。

所以，JWT 技术应运而生。

JWT，英文全称为 JSON Web Token，是为了在网络应用环境间传递声明而执行的一种基于 JSON 的开放标准。实际上就是在每次登录的时候通过一个 Token 字符串来校验登录状态。JWT 的声明一般被用来在身份提供者和服务提供者之间传递被认证的用户身份信息，以便于从资源服务器获取资源，也可以增加一些额外的业务逻辑所必须的声明信息，所以这个 Token 也可直接被用于认证，也可传递一些额外信息。

有了 JWT，一些认证就不需要借助于 Session 和 Cookie 了，服务器也无须维护 Session 信息，减少了服务器的开销。服务器只需要有一个校验 JWT 的功能就好了，同时也可以做到分布式部署和跨语言的支持。

JWT 通常就是一个加密的字符串，它也有自己的标准，类似下面的这种格式：

1

eyJ0eXAxIjoiMTIzNCIsImFsZzIiOiJhZG1pbiIsInR5cCI6IkpXVCIsImFsZyI6IkhTMjU2In0.eyJVc2VySWQiOjEyMywiVXNlck5hbWUiOiJhZG1pbiIsImV4cCI6MTU1MjI4Njc0Ni44Nzc0MDE4fQ.pEgdmFAy73walFonEm2zbxg46Oth3dlT02HR9iVzXa8

我们可以发现中间有两个用来分割的 . ，因此可以把它看成是一个三段式的加密字符串。

它由三部分构成，分别是 Header、Payload、Signature。

• Header，声明了 JWT 的签名算法，如 RSA、SHA256 等，也可能包含 JWT 编号或类型等数据，然后对整个信息进行 Base64 编码即可。
• Payload，通常用来存放一些业务需要但不敏感的信息，如 UserID 等，另外它也有很多默认是字段，如 JWT 签发者、JWT 接受者、JWT 过期时间等，Base64 编码即可。
• Signature，就是一个签名，是把 Header、Payload 的信息用秘钥 secret 加密后形成的，这个 secret 是保存在服务器端的，不能被轻易泄露。如此一来，即使一些 Payload 的信息被篡改，服务器也能通过 Signature 判断出非法请求，拒绝服务。

这三部分通过 . 组合起来就形成了 JWT 的字符串，就是用户的访问凭证。

所以这个登录认证流程也很简单了，用户拿着用户名密码登录，然后服务器生成 JWT 字符串返回给客户端。客户端每次请求都带着这个 JWT 就行了，服务器会自动判断其有效情况，如果有效，自然就返回对应的数据。JWT 的传输就多种多样了，可以将其放在 Request Headers 中，也可以放在 URL 里，甚至也有的网站把它放在 Cookie 里面，但总而言之，能传给服务器进行校验就好了。

好，到此为止呢，我们就已经了解了网站登录验证的实现了。

4. 模拟登录

好，那了解了网站登录验证的实现后，模拟登录自然就有思路了。

下面我们同样分两种认证方式来说明。

Session 和 Cookie

基于 Session 和 Cookie 的模拟登录，如果我们要用爬虫实现的话，其实最主要的就是把 Cookie 的信息维护好就行了，因为爬虫就相当于客户端浏览器，我们模拟好浏览器做的事情就好了。

一般怎么实现模拟登录呢？接下来我们结合之前所讲的技术总结一下。

• 第一，如果我们已经在浏览器中登录了自己的账号，要想用爬虫模拟，那么可以直接把 Cookie 复制过来交给爬虫。这是最省时省力的方式，相当于我们用浏览器手动操作登录了。我们把 Cookie 放到代码里，爬虫每次请求的时候再将其放到 Request Headers 中，完全模拟了浏览器的操作。之后服务器会通过 Cookie 校验登录状态，如果没问题，自然就可以执行某些操作或返回某些内容了。
• 第二，如果我们不想有任何手工操作，那么可以直接使用爬虫模拟登录过程。其实登录的过程多数也是一个 POST 请求。我们用爬虫提交了用户名、密码等信息给服务器，服务器返回的 Response Headers 里面可能会带有 Set-Cookie 的字段，我们只需要把这些 Cookie 保存下来就行了。所以，最主要的就是把这个过程中的 Cookie 维持好。当然这里可能会遇到一些困难，比如登录过程中伴随着各种校验参数，不好直接模拟请求；网站设置 Cookie 的过程是通过 JavaScript 实现的，所以可能还得仔细分析下其中的逻辑，尤其是我们用 requests 这样的请求库进行模拟登录的时候，遇到的问题经常比较多。
• 第三，我们也可以用一些简单的方式来实现模拟登录，即实现登录过程的自动化。比如我们用 Selenium、Pyppeteer 或 Playwright 来驱动浏览器模拟执行一些操作，如填写用户名和密码、提交表单等。登录成功后，通过 Selenium 或 Pyppeteer 获取当前浏览器的 Cookie 并保存即可。这样后续就可以拿着 Cookie 的内容发起请求，同样也能实现模拟登录。

以上介绍的就是一些常用的爬虫模拟登录的方案，其目的是维护好客户端的 Cookie 信息。总之，每次请求都携带好 Cookie 信息就能实现模拟登录了。

JWT

基于 JWT 的模拟登录思路也比较清晰了，由于 JWT 的字符串就是用户访问的凭证，所以模拟登录只需要做到下面几步。

• 第一步，模拟网站登录操作的请求。比如拿着用户名和密码信息请求登录接口，获取服务器返回的结果，这个结果中通常包含 JWT 字符串的信息，将它保存即可。
• 第二步，后续的请求携带 JWT 进行访问。在 JWT 不过期的情况下，通常能正常访问和执行对应的操作。携带方式多种多样，因网站而异。
• 第三步，如果 JWT 过期了，可能需要再次进行第一步，重新获取 JWT。

当然，模拟登录的过程肯定会带有一些其他的加密参数，需要根据实际情况具体分析。

4. 账号池

如果爬虫要求爬取的数据量比较大或爬取速度比较快，而网站又有单账号并发限制或者访问状态检测等反爬虫手段，那么我们的账号可能就会无法访问或者面临封号的风险了。

这时候一般怎么办呢？

我们可以使用分流的方案来实现。假设某个网站设置一分钟之内检测到同一个账号访问 3 次或 3 次以上则封号，我们就可以建立一个账号池，用多个账号来随机访问或爬取数据，这样就能大幅提高爬虫的并发量，降低被封号的风险了。比如我们可以准备 100 个账号，然后 100 个账号都模拟登录，把对应的 Cookie 或 JWT 存下来，每次访问的时候随机取一个来，由于账号多，所以每个账号被取用的概率也就降下来了，这样就能避免单账号并发过大的问题，也降低封号风险。

5. 总结

本节我们首先了解了 Session + Cookie 和 JWT 模拟登录的原理，接着初步了解了两种模拟登录方式的实现思路，最后初步介绍了一下账号池的作用。

后文我们会通过几个实战案例来实现上述两种方案的模拟登录，为了更好地理解后文的实战内容，建议好好理解本节所介绍的内容。

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3 网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

前面我们讲解了 Ajax 的分析方法，利用 Ajax 接口我们可以非常方便地完成数据爬取。只要我们能找到 Ajax 接口的规律，就可以通过某些参数构造出对应的请求，数据自然就能轻松爬取到。

但是在很多情况下，一些 Ajax 请求的接口通常会包含加密参数，如token、sign 等，如：https://spa2.scrape.center/，它的 Ajax 接口是包含一个 token 参数的，如图所示。

由于请求接口时必须加上 token 参数，所以我们如果不深入分析找到 token 的构造逻辑，是难以直接模拟这些 Ajax 请求的。

此时解决方法通常有两种：一种就是深挖其中的逻辑，把其中 token 的构造逻辑完全找出来，再用 Python 复现，构造 Ajax 请求；另外一种方法就是直接通过模拟浏览器的方式来绕过这个过程，因为在浏览器里我们可以看到这个数据，如果能把看到的数据直接爬取下来，当然也就能获取对应的信息了。

由于第一种方法难度较高，这里我们就先介绍第二种方法：模拟浏览器爬取。

这里使用的工具为 Selenium，这里就来先了解一下 Selenium 的基本使用方法。

Selenium 是一个自动化测试工具，利用它可以驱动浏览器执行特定的动作，如点击、下拉等操作，同时还可以获取浏览器当前呈现的页面的源代码，做到可见即可爬。对于一些 JavaScript 动态渲染的页面来说，此种抓取方式非常有效。本节中，就让我们来感受一下它的强大之处吧。

1. 准备工作

本节以 Chrome 为例来讲解 Selenium 的用法。在开始之前，请确保已经正确安装好了 Chrome 浏览器并配置好了 ChromeDriver。另外，还需要正确安装好 Python 的 Selenium 库。

安装方法可以参考：https://setup.scrape.center/selenium，全部配置完成之后，我们便可以开始本节的学习了。

2. 基本用法

准备工作做好之后，首先来大体看一下 Selenium 的功能。示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.wait import WebDriverWait

browser = webdriver.Chrome()
try:
    browser.get('https://www.baidu.com')
    input = browser.find_element_by_id('kw')
    input.send_keys('Python')
    input.send_keys(Keys.ENTER)
    wait = WebDriverWait(browser, 10)
    wait.until(EC.presence_of_element_located((By.ID, 'content_left')))
    print(browser.current_url)
    print(browser.get_cookies())
    print(browser.page_source)
finally:
    browser.close()

运行代码后发现，会自动弹出一个 Chrome 浏览器。浏览器首先会跳转到百度，然后在搜索框中输入 Python，接着跳转到搜索结果页，如图所示。

此时在控制台的输出结果如下：

1
2
3

https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=0&rsv_idx=1&tn=baidu&wd=Python&rsv_pq=c94d0df9000a72d0&rsv_t=07099xvun1ZmC0bf6eQvygJ43IUTTUOl5FCJVPgwG2YREs70GplJjH2F%2BCQ&rqlang=cn&rsv_enter=1&rsv_sug3=6&rsv_sug2=0&inputT=87&rsv_sug4=87
[{'secure': False, 'value': 'B490B5EBF6F3CD402E515D22BCDA1598', 'domain': '.baidu.com', 'path': '/', 'httpOnly': False, 'name': 'BDORZ', 'expiry': 1491688071.707553}, {'secure': False, 'value': '22473_1441_21084_17001', 'domain': '.baidu.com', 'path': '/', 'httpOnly': False, 'name': 'H_PS_PSSID'}, {'secure': False, 'value': '12883875381399993259_00_0_I_R_2_0303_C02F_N_I_I_0', 'domain': '.www.baidu.com', 'path': '/', 'httpOnly': False, 'name': '__bsi', 'expiry': 1491601676.69722}]
<!DOCTYPE html><!--STATUS OK-->...</html>

源代码过长，在此省略。可以看到，我们得到的当前 URL、Cookies 和源代码都是浏览器中的真实内容。

所以说，如果用 Selenium 来驱动浏览器加载网页的话，就可以直接拿到 JavaScript 渲染的结果了，不用担心使用的是什么加密系统。

下面来详细了解一下 Selenium 的用法。

3. 声明浏览器对象

Selenium 支持非常多的浏览器，如 Chrome、Firefox、Edge 等，还有 Android、BlackBerry 等手机端的浏览器。我们可以用如下方式初始化：

1
2
3
4
5
6

from selenium import webdriver

browser = webdriver.Chrome()
browser = webdriver.Firefox()
browser = webdriver.Edge()
browser = webdriver.Safari()

这样就完成了浏览器对象的初始化并将其赋值为 browser 对象。接下来，我们要做的就是调用 browser 对象，让其执行各个动作以模拟浏览器操作。

4. 访问页面

我们可以用 get 方法来请求网页，其参数传入链接 URL 即可。比如，这里用 get 方法访问淘宝，然后打印出源代码，代码如下：

1
2
3
4
5
6

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
print(browser.page_source)
browser.close()

运行后发现，此时弹出了 Chrome 浏览器并且自动访问了淘宝，然后控制台输出了淘宝页面的源代码，随后浏览器关闭。

通过这几行简单的代码，我们可以实现浏览器的驱动并获取网页源码，非常便捷。

5. 查找节点

Selenium 可以驱动浏览器完成各种操作，比如填充表单、模拟点击等。比如，我们想要完成向某个输入框输入文字的操作，总需要知道这个输入框在哪里吧？而 Selenium 提供了一系列查找节点的方法，我们可以用这些方法来获取想要的节点，以便下一步执行一些动作或者提取信息。

单个节点

比如，想要从淘宝页面中提取搜索框这个节点，首先要观察它的源代码，如图所示。

可以发现，它的 id 是 q，name 也是 q。此外，还有许多其他属性，此时我们就可以用多种方式获取它了。比如，find_element_by_name 是根据 name 值获取，find_element_by_id 是根据 id 获取。另外，还有根据 XPath、CSS 选择器等获取的方式。

下面我们用代码实现一下：

1
2
3
4
5
6
7
8
9

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
input_first = browser.find_element_by_id('q')
input_second = browser.find_element_by_css_selector('#q')
input_third = browser.find_element_by_xpath('//*[@id="q"]')
print(input_first, input_second, input_third)
browser.close()

这里我们使用 3 种方式获取输入框，分别是根据 ID、CSS 选择器和 XPath 获取，它们返回的结果完全一致。运行结果如下：

1
2
3

<selenium.webdriver.remote.webelement.WebElement (session="5e53d9e1c8646e44c14c1c2880d424af", element="0.5649563096161541-1")>
<selenium.webdriver.remote.webelement.WebElement (session="5e53d9e1c8646e44c14c1c2880d424af", element="0.5649563096161541-1")>
<selenium.webdriver.remote.webelement.WebElement (session="5e53d9e1c8646e44c14c1c2880d424af", element="0.5649563096161541-1")>

可以看到，这 3 个节点都是 WebElement 类型，是完全一致的。

下面列出所有获取单个节点的方法：

1
2
3
4
5
6
7
8

find_element_by_id
find_element_by_name
find_element_by_xpath
find_element_by_link_text
find_element_by_partial_link_text
find_element_by_tag_name
find_element_by_class_name
find_element_by_css_selector

另外，Selenium 还提供了通用方法 find_element，它需要传入两个参数：查找方式 By 和值。实际上，它就是 find_element_by_id 这种方法的通用函数版本，比如 find_element_by_id(id) 就等价于 find_element(By.ID, id)，二者得到的结果完全一致。我们用代码实现一下：

1
2
3
4
5
6
7
8

from selenium import webdriver
from selenium.webdriver.common.by import By

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
input_first = browser.find_element(By.ID, 'q')
print(input_first)
browser.close()

实际上，这种查找方式的功能和上面列举的查找函数完全一致，不过参数更加灵活。

多个节点

如果查找的目标在网页中只有一个，那么完全可以用 find_element 方法。但如果有多个节点，再用 find_element 方法查找，就只能得到第一个节点了。如果要查找所有满足条件的节点，需要用 find_elements 这样的方法。注意，在这个方法的名称中，element 多了一个 s，注意区分。

比如，要查找淘宝左侧导航条的所有条目，就可以这样来实现：

1
2
3
4
5
6
7

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
lis = browser.find_elements_by_css_selector('.service-bd li')
print(lis)
browser.close()

运行结果如下：

1

[<selenium.webdriver.remote.webelement.WebElement (session="c26290835d4457ebf7d96bfab3740d19", element="0.09221044033125603-1")>, <selenium.webdriver.remote.webelement.WebElement (session="c26290835d4457ebf7d96bfab3740d19", element="0.09221044033125603-2")>, <selenium.webdriver.remote.webelement.WebElement (session="c26290835d4457ebf7d96bfab3740d19", element="0.09221044033125603-3")>...<selenium.webdriver.remote.webelement.WebElement (session="c26290835d4457ebf7d96bfab3740d19", element="0.09221044033125603-16")>]

这里简化了输出结果，中间部分省略。

可以看到，得到的内容变成了列表类型，列表中的每个节点都是 WebElement 类型。

也就是说，如果我们用 find_element 方法，只能获取匹配的第一个节点，结果是 WebElement 类型。如果用 find_elements 方法，则结果是列表类型，列表中的每个节点都是 WebElement 类型。

这里列出所有获取多个节点的方法：

1
2
3
4
5
6
7
8

find_elements_by_id
find_elements_by_name
find_elements_by_xpath
find_elements_by_link_text
find_elements_by_partial_link_text
find_elements_by_tag_name
find_elements_by_class_name
find_elements_by_css_selector

当然，我们也可以直接用 find_elements 方法来选择，这时可以这样写：

1

lis = browser.find_elements(By.CSS_SELECTOR, '.service-bd li')

结果是完全一致的。

6. 节点交互

Selenium 可以驱动浏览器来执行一些操作，也就是说可以让浏览器模拟执行一些动作。比较常见的用法有：输入文字时用 send_keys 方法，清空文字时用 clear 方法，点击按钮时用 click 方法。示例如下：

1
2
3
4
5
6
7
8
9
10
11
12

from selenium import webdriver
import time

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
input = browser.find_element_by_id('q')
input.send_keys('iPhone')
time.sleep(1)
input.clear()
input.send_keys('iPad')
button = browser.find_element_by_class_name('btn-search')
button.click()

这里首先驱动浏览器打开淘宝，然后用 find_element_by_id 方法获取输入框，然后用 send_keys 方法输入 iPhone 文字，等待一秒后用 clear 方法清空输入框，再次调用 send_keys 方法输入 iPad 文字，之后再用 find_element_by_class_name 方法获取搜索按钮，最后调用 click 方法完成搜索动作。

通过上面的方法，我们完成了一些常见节点的操作，更多的操作可以参见官方文档的交互动作介绍 ：http://selenium-python.readthedocs.io/api.html#module-selenium.webdriver.remote.webelement。

7. 动作链

在上面的实例中，一些交互动作都是针对某个节点执行的。比如，对于输入框，我们就调用它的输入文字和清空文字方法；对于按钮，就调用它的点击方法。其实，还有另外一些操作，它们没有特定的执行对象，比如鼠标拖曳、键盘按键等，这些动作用另一种方式来执行，那就是动作链。

比如，现在实现一个节点的拖曳操作，将某个节点从一处拖曳到另外一处，可以这样实现：

1
2
3
4
5
6
7
8
9
10
11
12

from selenium import webdriver
from selenium.webdriver import ActionChains

browser = webdriver.Chrome()
url = 'http://www.runoob.com/try/try.php?filename=jqueryui-api-droppable'
browser.get(url)
browser.switch_to.frame('iframeResult')
source = browser.find_element_by_css_selector('#draggable')
target = browser.find_element_by_css_selector('#droppable')
actions = ActionChains(browser)
actions.drag_and_drop(source, target)
actions.perform()

首先，打开网页中的一个拖曳实例，然后依次选中要拖曳的节点和拖曳到的目标节点，接着声明 ActionChains 对象并将其赋值为 actions 变量，然后通过调用 actions 变量的 drag_and_drop 方法，再调用 perform 方法执行动作，此时就完成了拖曳操作，如图所示。

更多的动作链操作可以参考官方文档的动作链介绍：http://selenium-python.readthedocs.io/api.html#module-selenium.webdriver.common.action_chains。

8. 执行 JavaScript

对于某些操作，Selenium API 并没有提供。比如，下拉进度条，它可以直接模拟运行 JavaScript，此时使用 execute_script 方法即可实现，代码如下：

1
2
3
4
5
6

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.zhihu.com/explore')
browser.execute_script('window.scrollTo(0, document.body.scrollHeight)')
browser.execute_script('alert("To Bottom")')

这里就利用 execute_script 方法将进度条下拉到最底部，然后弹出 alert 提示框。

所以说有了这个方法，基本上 API 没有提供的所有功能都可以用执行 JavaScript 的方式来实现了。

9. 获取节点信息

前面说过，通过 page_source 属性可以获取网页的源代码，接着就可以使用解析库（如正则表达式、Beautiful Soup、pyquery 等）来提取信息了。

不过，既然 Selenium 已经提供了选择节点的方法，返回的是 WebElement 类型，那么它也有相关的方法和属性来直接提取节点信息，如属性、文本等。这样的话，我们就可以不用通过解析源代码来提取信息了，非常方便。

接下来，我们就来看看怎样获取节点信息吧。

获取属性

我们可以使用 get_attribute 方法来获取节点的属性，但是其前提是先选中这个节点，示例如下：

1
2
3
4
5
6
7
8

from selenium import webdriver

browser = webdriver.Chrome()
url = 'https://spa2.scrape.center/'
browser.get(url)
logo = browser.find_element_by_class_name('logo-image')
print(logo)
print(logo.get_attribute('src'))

运行之后，程序便会驱动浏览器打开该页面，然后获取 class 为 logo-image 的节点，最后打印出它的 src。

控制台的输出结果如下：

1
2

<selenium.webdriver.remote.webelement.WebElement (session="7f4745d35a104759239b53f68a6f27d0", element="cd7c72b4-4920-47ed-91c5-ea06601dc509")>
https://spa2.scrape.center/img/logo.a508a8f0.png

通过 get_attribute 方法，然后传入想要获取的属性名，就可以得到它的值了。

获取文本值

每个 WebElement 节点都有 text 属性，直接调用这个属性就可以得到节点内部的文本信息，这相当于 pyquery 的 text 方法，示例如下：

1
2
3
4
5
6
7

from selenium import webdriver

browser = webdriver.Chrome()
url = 'https://spa2.scrape.center/'
browser.get(url)
input = browser.find_element_by_class_name('logo-title')
print(input.text)

这里依然先打开页面，然后获取 class 为 logo-title 这个节点，再将其文本值打印出来。

控制台的输出结果如下：

1

Scrape

获取 ID、位置、标签名和大小

另外，WebElement 节点还有一些其他属性，比如 id 属性可以获取节点 ID，location 属性可以获取该节点在页面中的相对位置，tag_name 属性可以获取标签名称，size 属性可以获取节点的大小，也就是宽高，这些属性有时候还是很有用的。示例如下：

1
2
3
4
5
6
7
8
9
10

from selenium import webdriver

browser = webdriver.Chrome()
url = 'https://spa2.scrape.center/'
browser.get(url)
input = browser.find_element_by_class_name('logo-title')
print(input.id)
print(input.location)
print(input.tag_name)
print(input.size)

这里首先获得 class 为 logo-title 这个节点，然后调用其 id、location、tag_name、size 属性来获取对应的属性值。

10. 切换 Frame

我们知道网页中有一种节点叫作 iframe，也就是子 Frame，相当于页面的子页面，它的结构和外部网页的结构完全一致。Selenium 打开页面后，它默认是在父级 Frame 里面操作，而此时如果页面中还有子 Frame，它是不能获取到子 Frame 里面的节点的。这时就需要使用 switch_to.frame 方法来切换 Frame。示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import time
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException

browser = webdriver.Chrome()
url = 'http://www.runoob.com/try/try.php?filename=jqueryui-api-droppable'
browser.get(url)
browser.switch_to.frame('iframeResult')
try:
    logo = browser.find_element_by_class_name('logo')
except NoSuchElementException:
    print('NO LOGO')
browser.switch_to.parent_frame()
logo = browser.find_element_by_class_name('logo')
print(logo)
print(logo.text)

控制台输出结果如下：

1
2
3

NO LOGO
<selenium.webdriver.remote.webelement.WebElement (session="4bb8ac03ced4ecbdefef03ffdc0e4ccd", element="0.13792611320464965-2")>
RUNOOB.COM

这里还是以前面演示动作链操作的网页为实例，首先通过 switch_to.frame 方法切换到子 Frame 里面，然后尝试获取子 Frame 里的 logo 节点（这是找不到的），如果找不到的话，就会抛出 NoSuchElementException 异常，异常被捕捉之后，就会输出 NO LOGO。接下来，重新切换回父级 Frame，然后再次重新获取节点，发现此时可以成功获取了。

所以，当页面中包含子 Frame 时，如果想获取子 Frame 中的节点，需要先调用 switch_to.frame 方法切换到对应的 Frame，然后再进行操作。

11. 延时等待

在 Selenium 中，get 方法会在网页框架加载结束后结束执行，此时如果获取 page_source，可能并不是浏览器完全加载完成的页面，如果某些页面有额外的 Ajax 请求，我们在网页源代码中也不一定能成功获取到。所以，这里需要延时等待一定时间，确保节点已经加载出来。

这里等待方式有两种：一种是隐式等待，一种是显式等待。

隐式等待

当使用隐式等待执行测试的时候，如果 Selenium 没有在 DOM 中找到节点，将继续等待，超出设定时间后，则抛出找不到节点的异常。换句话说，当查找节点而节点并没有立即出现的时候，隐式等待将等待一段时间再查找 DOM，默认的时间是 0。示例如下：

1
2
3
4
5
6
7

from selenium import webdriver

browser = webdriver.Chrome()
browser.implicitly_wait(10)
browser.get('https://spa2.scrape.center/')
input = browser.find_element_by_class_name('logo-image')
print(input)

这里我们用 implicitly_wait 方法实现了隐式等待。

显式等待

隐式等待的效果其实并没有那么好，因为我们只规定了一个固定时间，而页面的加载时间会受到网络条件的影响。

这里还有一种更合适的显式等待方法，它指定要查找的节点，然后指定一个最长等待时间。如果在规定时间内加载出来了这个节点，就返回查找的节点；如果到了规定时间依然没有加载出该节点，则抛出超时异常。示例如下：

1
2
3
4
5
6
7
8
9
10
11

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

browser = webdriver.Chrome()
browser.get('https://www.taobao.com/')
wait = WebDriverWait(browser, 10)
input = wait.until(EC.presence_of_element_located((By.ID, 'q')))
button = wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, '.btn-search')))
print(input, button)

这里首先引入 WebDriverWait 这个对象，指定最长等待时间，然后调用它的 until 方法，传入等待条件 expected_conditions。比如，这里传入了 presence_of_element_located 这个条件，代表节点出现的意思，其参数是节点的定位元组，也就是 ID 为 q 的节点搜索框。

这样可以做到的效果就是，在 10 秒内如果 ID 为 q 的节点（即搜索框）成功加载出来，就返回该节点；如果超过 10 秒还没有加载出来，就抛出异常。

对于按钮，可以更改一下等待条件，比如改为 element_to_be_clickable，也就是可点击，所以查找按钮时查找 CSS 选择器为 .btn-search 的按钮，如果 10 秒内它是可点击的，也就是成功加载出来了，就返回这个按钮节点；如果超过 10 秒还不可点击，也就是没有加载出来，就抛出异常。

运行代码，在网速较佳的情况下是可以成功加载出来的。

控制台的输出如下：

1
2

<selenium.webdriver.remote.webelement.WebElement (session="07dd2fbc2d5b1ce40e82b9754aba8fa8", element="0.5642646294074107-1")>
<selenium.webdriver.remote.webelement.WebElement (session="07dd2fbc2d5b1ce40e82b9754aba8fa8", element="0.5642646294074107-2")>

可以看到，控制台成功输出了两个节点，它们都是 WebElement 类型。

如果网络有问题，10 秒内没有成功加载，那就抛出 TimeoutException 异常，此时控制台的输出如下：

1
2
3
4
5

TimeoutException Traceback (most recent call last)
<ipython-input-4-f3d73973b223> in <module>()
      7 browser.get('https://www.taobao.com/')
      8 wait = WebDriverWait(browser, 10)
----> 9 input = wait.until(EC.presence_of_element_located((By.ID, 'q')))

关于等待条件，其实还有很多，比如判断标题内容，判断某个节点内是否出现了某文字等。下表列出了所有的等待条件。

更多等待条件的参数及用法介绍可以参考官方文档：http://selenium-python.readthedocs.io/api.html#module-selenium.webdriver.support.expected_conditions。

12. 前进后退

平常使用浏览器时，都有前进和后退功能，Selenium 也可以完成这个操作，它使用 back 方法后退，使用 forward 方法前进。示例如下：

1
2
3
4
5
6
7
8
9
10
11

import time
from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.baidu.com/')
browser.get('https://www.taobao.com/')
browser.get('https://www.python.org/')
browser.back()
time.sleep(1)
browser.forward()
browser.close()

这里我们连续访问 3 个页面，然后调用 back 方法回到第二个页面，接下来再调用 forward 方法又可以前进到第三个页面。

13. Cookies

使用 Selenium，还可以方便地对 Cookies 进行操作，例如获取、添加、删除 Cookies 等。示例如下：

1
2
3
4
5
6
7
8
9

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.zhihu.com/explore')
print(browser.get_cookies())
browser.add_cookie({'name': 'name', 'domain': 'www.zhihu.com', 'value': 'germey'})
print(browser.get_cookies())
browser.delete_all_cookies()
print(browser.get_cookies())

首先，我们访问了知乎。加载完成后，浏览器实际上已经生成 Cookies 了。接着，调用 get_cookies 方法获取所有的 Cookies。然后，我们添加一个 Cookie，这里传入一个字典，有 name、domain 和 value 等内容。接下来，再次获取所有的 Cookies。可以发现，结果就多了这一项新加的 Cookie。最后，调用 delete_all_cookies 方法删除所有的 Cookies。再重新获取，发现结果就为空了。

控制台的输出如下：

1
2
3

[{'secure': False, 'value': '"NGM0ZTM5NDAwMWEyNDQwNDk5ODlkZWY3OTkxY2I0NDY=|1491604091|236e34290a6f407bfbb517888849ea509ac366d0"', 'domain': '.zhihu.com', 'path': '/', 'httpOnly': False, 'name': 'l_cap_id', 'expiry': 1494196091.403418}, ...]
[{'secure': False, 'value': 'germey', 'domain': '.www.zhihu.com', 'path': '/', 'httpOnly': False, 'name': 'name'}, {'secure': False, 'value': '"NGM0ZTM5NDAwMWEyNDQwNDk5ODlkZWY3OTkxY2I0NDY=|1491604091|236e34290a6f407bfbb517888849ea509ac366d0"', 'domain': '.zhihu.com', 'path': '/', 'httpOnly': False, 'name': 'l_cap_id', 'expiry': 1494196091.403418}, ...]
[]

通过以上方法来操作 Cookies 还是非常方便的。

14. 选项卡管理

在访问网页的时候，会开启一个个选项卡。在 Selenium 中，我们也可以对选项卡进行操作。示例如下：

1
2
3
4
5
6
7
8
9
10
11
12

import time
from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.baidu.com')
browser.execute_script('window.open()')
print(browser.window_handles)
browser.switch_to.window(browser.window_handles[1])
browser.get('https://www.taobao.com')
time.sleep(1)
browser.switch_to.window(browser.window_handles[0])
browser.get('https://python.org')

控制台的输出如下：

1

['CDwindow-4f58e3a7-7167-4587-bedf-9cd8c867f435', 'CDwindow-6e05f076-6d77-453a-a36c-32baacc447df']

这里首先访问了百度，然后调用了 execute_script 方法，这里传入 window.open 这个 JavaScript 语句新开启一个选项卡。接下来，我们想切换到该选项卡。这里调用 window_handles 属性获取当前开启的所有选项卡，返回的是选项卡的代号列表。要想切换选项卡，只需要调用 switch_to.window 方法即可，其中参数是选项卡的代号。这里我们将第二个选项卡代号传入，即跳转到第二个选项卡，接下来在第二个选项卡下打开一个新页面，然后切换回第一个选项卡重新调用 switch_to.window 方法，再执行其他操作即可。

15. 异常处理

在使用 Selenium 的过程中，难免会遇到一些异常，例如超时、节点未找到等错误，一旦出现此类错误，程序便不会继续运行了。这里我们可以使用 try except 语句来捕获各种异常。

首先，演示一下节点未找到的异常，示例如下：

1
2
3
4
5

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.baidu.com')
browser.find_element_by_id('hello')

这里首先打开百度页面，然后尝试选择一个并不存在的节点，此时就会遇到异常。

运行之后控制台的输出如下：

1
2
3
4
5

NoSuchElementException Traceback (most recent call last)
<ipython-input-23-978945848a1b> in <module>()
      3 browser = webdriver.Chrome()
      4 browser.get('https://www.baidu.com')
----> 5 browser.find_element_by_id('hello')

可以看到，这里抛出了 NoSuchElementException 异常，这通常是节点未找到的异常。为了防止程序遇到异常而中断，我们需要捕获这些异常，示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

from selenium import webdriver
from selenium.common.exceptions import TimeoutException, NoSuchElementException

browser = webdriver.Chrome()
try:
    browser.get('https://www.baidu.com')
except TimeoutException:
    print('Time Out')
try:
    browser.find_element_by_id('hello')
except NoSuchElementException:
    print('No Element')
finally:
    browser.close()

这里我们使用 try except 来捕获各类异常。比如，我们对 find_element_by_id 查找节点的方法捕获 NoSuchElementException 异常，这样一旦出现这样的错误，就进行异常处理，程序也不会中断了。

控制台的输出如下：

1

No Element

关于更多的异常类，可以参考官方文档：：http://selenium-python.readthedocs.io/api.html#module-selenium.common.exceptions。

16. 反屏蔽

现在很多网站都加上了对 Selenium 的检测，来防止一些爬虫的恶意爬取。即如果检测到有人在使用 Selenium 打开浏览器，那就直接屏蔽。

在大多数情况下，检测的基本原理是检测当前浏览器窗口下的 window.navigator 对象是否包含 webdriver 这个属性。因为在正常使用浏览器的情况下，这个属性是 undefined，然而一旦我们使用了 Selenium，Selenium 会给 window.navigator 设置 webdriver 属性。很多网站就通过 JavaScript 判断如果 webdriver 属性存在，那就直接屏蔽。

这边有一个典型的案例网站：https://antispider1.scrape.center/，这个网站就使用了上述原理实现了 WebDriver 的检测，如果使用 Selenium 直接爬取的话，那就会返回如图所示的页面。

这时候我们可能想到直接使用 JavaScript 语句把这个 webdriver 属性置空，比如通过调用 execute_script 方法来执行如下代码：

1

Object.defineProperty(navigator, "webdriver", { get: () => undefined });

这行 JavaScript 语句的确可以把 webdriver 属性置空，但是 execute_script 调用这行 JavaScript 语句实际上是在页面加载完毕之后才执行的，执行太晚了，网站早在最初页面渲染之前就已经对 webdriver 属性进行了检测，所以用上述方法并不能达到效果。

在 Selenium 中，我们可以使用 CDP（即 Chrome Devtools-Protocol，Chrome 开发工具协议）来解决这个问题，通过它我们可以实现在每个页面刚加载的时候执行 JavaScript 代码，执行的 CDP 方法叫作 Page.addScriptToEvaluateOnNewDocument，然后传入上文的 JavaScript 代码即可，这样我们就可以在每次页面加载之前将 webdriver 属性置空了。另外，我们还可以加入几个选项来隐藏 WebDriver 提示条和自动化扩展信息，代码实现如下：

1
2
3
4
5
6
7
8
9
10
11

from selenium import webdriver
from selenium.webdriver import ChromeOptions

option = ChromeOptions()
option.add_experimental_option('excludeSwitches', ['enable-automation'])
option.add_experimental_option('useAutomationExtension', False)
browser = webdriver.Chrome(options=option)
browser.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {
    'source': 'Object.defineProperty(navigator, "webdriver", {get: () => undefined})'
})
browser.get('https://antispider1.scrape.center/')

这样整个页面就能被加载出来了，如图所示。

对于大多数情况，以上方法均可以实现 Selenium 反屏蔽。但对于一些特殊网站，如果它有更多的 WebDriver 特征检测，可能需要具体排查。

17. 无头模式

我们可以观察到，上面的案例在运行的时候，总会弹出一个浏览器窗口，虽然有助于观察页面爬取状况，但在有时候窗口弹来弹去也会形成一些干扰。

Chrome 浏览器从 60 版本已经支持了无头模式，即 Headless。无头模式在运行的时候不会再弹出浏览器窗口，减少了干扰，而且它减少了一些资源的加载，如图片等，所以也在一定程度上节省了资源加载时间和网络带宽。

我们可以借助于 ChromeOptions 来开启 Chrome Headless 模式，代码实现如下：

1
2
3
4
5
6
7
8
9

from selenium import webdriver
from selenium.webdriver import ChromeOptions

option = ChromeOptions()
option.add_argument('--headless')
browser = webdriver.Chrome(options=option)
browser.set_window_size(1366, 768)
browser.get('https://www.baidu.com')
browser.get_screenshot_as_file('preview.png')

这里我们通过 ChromeOptions 的 add_argument 方法添加了一个参数 --headless，开启了无头模式。在无头模式下，我们最好设置一下窗口的大小，接着打开页面，最后我们调用 get_screenshot_as_file 方法输出了页面的截图。

运行代码之后，我们发现 Chrome 窗口就不会再弹出来了，代码依然正常运行，最后输出的页面如图所示。

这样我们就在无头模式下完成了页面的抓取和截图操作。

18. 总结

现在，我们基本上对 Selenium 的常规用法有了大体的了解。使用 Selenium，处理 JavaScript 渲染的页面不再是难事，后面我们会用一个实例来演示 Selenium 爬取网站的流程。

本节代码：https://github.com/Python3WebSpider/SeleniumTest。

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

想查数据，就免不了搜索，而搜索离不开搜索引擎。百度、谷歌都是非常庞大、复杂的搜索引擎，它们几乎索引了互联网上开放的所有网页和数据。然而对于我们自己的业务数据来说，没必要用这么复杂的技术。如果我们想实现自己的搜索引擎，为了便于存储和检索，Elasticsearch 就是不二选择。它是一个全文搜索引擎，可以快速存储、搜索和分析海量数据。

所以，如果我们我们将爬取到的数据存储到 Elasticsearch 里面，那将会非常方便检索。

1. Elasticsearch 介绍

Elasticsearch 是一个开源的搜索引擎，建立在一个全文搜索引擎库 Apache Lucene™ 基础之上。

那 Lucene 又是什么呢？Lucene 可能是目前存在的（不论开源还是私有的）拥有最先进、高性能和全功能搜索引擎功能的库，但也仅仅只是一个库。要想用 Lucene，我们需要编写 Java 并引用 Lucene 包才可以，而且我们需要对信息检索有一定程度的理解。

为了解决这个问题，Elasticsearch 就诞生了。Elasticsearch 也是使用 Java 编写的，它的内部使用 Lucene 做索引与搜索，但是它的目标是使全文检索变得简单，相当于 Lucene 的一层封装，它提供了一套简单一致的 RESTful API 来帮助我们实现存储和检索。

所以 Elasticsearch 仅仅就是一个简易版的 Lucene 封装吗？那就大错特错了，Elasticsearch 不仅仅是 Lucene，并且也不仅仅只是一个全文搜索引擎。它可以这样准确形容：

• 一个分布式的实时文档存储，每个字段可以被索引与搜索；
• 一个分布式实时分析搜索引擎；
• 能胜任上百个服务节点的扩展，并支持 PB 级别的结构化或者非结构化数据。

总之，它是一个非常强大的搜索引擎，维基百科、Stack Overflow、GitHub 都纷纷采用它来做搜索，不仅仅提供强大的检索能力，也提供强大的存储能力。

2. Elasticsearch 相关概念

在 Elasticsearch 中有几个基本概念，如节点、索引、文档等，下面分别说明一下。理解了这些概念，对熟悉 Elasticsearch 是非常有帮助的。

节点和集群

Elasticsearch 本质上是一个分布式数据库，允许多台服务器协同工作，每台服务器可以运行多个 Elasticsearch 实例。

单个 Elasticsearch 实例称为一个节点（Node），一组节点构成一个集群（Cluster）。

索引

索引，即 Index，Elasticsearch 会索引所有字段，经过处理后写入一个反向索引（Inverted Index）。查找数据的时候，直接查找该索引。

所以，Elasticsearch 数据管理的顶层单位就叫作索引，其实就相当于 MySQL、MongoDB 等中数据库的概念。另外，值得注意的是，每个索引 （即数据库）的名字必须小写。

文档

文档，即 Document。索引里面单条记录称为文档，许多条文档构成了一个索引。

同一个索引里面的文档，不要求有相同的结构（Schema），但是最好保持一致，因为这样有利于提高搜索效率。

类型

文档可以分组，比如 weather 这个索引里面，既可以按城市分组（北京和上海），也可以按气候分组（晴天和雨天）。这种分组就叫作类型（Type），它是虚拟的逻辑分组，用来过滤文档，类似 MySQL 中的数据表、MongoDB 中的 Collection。

不同的类型应该有相似的结构。举例来说，id 字段不能在这个组中是字符串，在另一个组中是数值。这是与关系型数据库的表的一个区别。性质完全不同的数据（比如 products 和 logs）应该存成两个索引，而不是一个索引里面的两个类型（虽然可以做到）。

根据规划，Elastic 6.x 版只允许每个索引包含一个类型，Elastic 7.x 开始将会将其彻底移除。

字段

每个文档都类似一个 JSON 结构，它包含了许多字段，每个字段都有其对应的值，多个字段组成了一个文档，其实就可以类比 MySQL 数据表中的字段。

在 Elasticsearch 中，文档归属于一种类型（Type），而这些类型存在于索引中，我们可以画一些简单的对比图来类比传统关系型数据库：

1
2

Relational DB -> Databases -> Tables -> Rows -> Columns
Elasticsearch -> Indices   -> Types  -> Documents -> Fields

以上就是 Elasticsearch 里面的一些基本概念，通过和关系型数据库的对比更加有助于理解。

3. 准备工作

在开始本节实际操作之前，请确保已经正确安装好了 Elasticsearch，安装方式可以参考：https://setup.scrape.center/elasticsearch，安装完成之后确保其在本地 9200 端口上正常运行即可。

Elasticsearch 实际上提供了一系列 Restful API 来进行存取和查询操作，我们可以使用 curl 等命令或者直接调用 API 来进行数据存储和修改操作，但总归来说并不是很方便。所以这里我们就直接介绍一个专门用来对接 Elasticsearch 操作的 Python 库，名称也叫做 Elasticsearch，使用 pip3 安装即可：

1

pip3 install elasticsearch

更详细的安装方式可以参考：https://setup.scrape.center/elasticsearch-py。

安装好了之后我们就可以开始本节的学习了。

4. 创建索引

我们先来看下怎样创建一个索引，这里我们创建一个名为 news 的索引：

1
2
3
4
5

from elasticsearch import Elasticsearch

es = Elasticsearch()
result = es.indices.create(index='news', ignore=400)
print(result)

这里我们首先创建了一个 Elasticsearch 对象，并且没有设置任何参数，默认情况下它会连接本地 9200 端口运行的 Elasticsearch 服务，我们也可以设置特定的连接信息，如：

1
2
3
4

es = Elasticsearch(
    ['https://[username:password@]hostname:port'],
    verify_certs=True, # 是否验证 SSL 证书
)

第一个参数我们可以构造特定格式的链接字符串并传入，hostname 和 port 即 Elasticsearch 运行的地址和端口，username 和 password 是可选的，代表连接 Elasticsearch 需要的用户名和密码，另外而且还有其他的参数设置，比如 verify_certs 代表是否验证证书有效性。更多参数的设置可以参考：https://elasticsearch-py.readthedocs.io/en/latest/api.html#elasticsearch。

声明 Elasticsearch 对象之后，我们调用了 es 的 indices 对象的 create 方法传入了 index 的名称，如果创建成功，会返回如下结果：

1

{'acknowledged': True, 'shards_acknowledged': True, 'index': 'news'}

可以看到，其返回结果是 JSON 格式，其中的 acknowledged 字段表示创建操作执行成功。

但这时如果我们再把代码执行一次的话，就会返回如下结果：

1

{'error': {'root_cause': [{'type': 'resource_already_exists_exception', 'reason': 'index [news/hHEYozoqTzK_qRvV4j4a3w] already exists', 'index_uuid': 'hHEYozoqTzK_qRvV4j4a3w', 'index': 'news'}], 'type': 'resource_already_exists_exception', 'reason': 'index [news/hHEYozoqTzK_qRvV4j4a3w] already exists', 'index_uuid': 'hHEYozoqTzK_qRvV4j4a3w', 'index': 'news'}, 'status': 400}

它提示创建失败，status 状态码是 400，错误原因是索引已经存在了。

注意在这里的代码中，我们使用的 ignore 参数为 400，这说明如果返回结果是 400 的话，就忽略这个错误，不会报错，程序不会抛出异常。

假如我们不加 ignore 这个参数的话：

1
2
3

es = Elasticsearch()
result = es.indices.create(index='news')
print(result)

再次执行就会报错了：

1
2

raise HTTP_EXCEPTIONS.get(status_code, TransportError)(status_code, error_message, additional_info)
elasticsearch.exceptions.RequestError: TransportError(400, 'resource_already_exists_exception', 'index [news/QM6yz2W8QE-bflKhc5oThw] already exists')

这样程序的执行就会出现问题。因此，我们需要善用 ignore 参数，把一些意外情况排除，这样可以保证程序正常执行而不会中断。

创建完之后，我们还可以设置下索引的字段映射定义，可以参考：https://elasticsearch-py.readthedocs.io/en/latest/api.html?#elasticsearch.client.IndicesClient.put_mapping。

5. 删除索引

删除索引也是类似的，代码如下：

1
2
3
4
5

from elasticsearch import Elasticsearch

es = Elasticsearch()
result = es.indices.delete(index='news', ignore=[400, 404])
print(result)

这里也使用了 ignore 参数来忽略索引不存在而删除失败导致程序中断的问题。

如果删除成功，会输出如下结果：

1

{'acknowledged': True}

如果索引已经被删除，再执行删除，则会输出如下结果：

1

{'error': {'root_cause': [{'type': 'index_not_found_exception', 'reason': 'no such index [news]', 'resource.type': 'index_or_alias', 'resource.id': 'news', 'index_uuid': '_na_', 'index': 'news'}], 'type': 'index_not_found_exception', 'reason': 'no such index [news]', 'resource.type': 'index_or_alias', 'resource.id': 'news', 'index_uuid': '_na_', 'index': 'news'}, 'status': 404}

这个结果表明当前索引不存在，删除失败。返回的结果同样是 JSON，状态码是 404，但是由于我们添加了 ignore 参数，忽略了 404 状态码，因此程序正常执行，输出 JSON 结果，而不是抛出异常。

6. 插入数据

Elasticsearch 就像 MongoDB 一样，在插入数据的时候可以直接插入结构化字典数据，插入数据可以调用 create 方法。例如，这里我们插入一条新闻数据：

1
2
3
4
5
6
7
8
9
10
11

from elasticsearch import Elasticsearch

es = Elasticsearch()
es.indices.create(index='news', ignore=400)

data = {
  'title': '乘风破浪不负韶华，奋斗青春圆梦高考',
  'url': 'http://view.inews.qq.com/a/EDU2021041600732200'
}
result = es.create(index='news', id=1, body=data)
print(result)

这里我们首先声明了一条新闻数据，包括标题和链接，然后通过调用 create 方法插入了这条数据。在调用 create 方法时，我们传入了 4 个参数，index 参数代表了索引名称，id 则是数据的唯一标识 ID，body 则代表了文档的具体内容。

运行结果如下：

1

{'_index': 'news', '_type': '_doc', '_id': '1', '_version': 1, 'result': 'created', '_shards': {'total': 2, 'successful': 1, 'failed': 0}, '_seq_no': 0, '_primary_term': 1}

结果中 result 字段为 created，代表该数据插入成功。

另外，其实我们也可以使用 index 方法来插入数据。但与 create 不同的是，create 方法需要我们指定 id 字段来唯一标识该条数据，而 index 方法则不需要，如果不指定 id，会自动生成一个 id。调用 index 方法的写法如下：

1

es.index(index='news', body=data)

create 方法内部其实也是调用了 index 方法，是对 index 方法的封装。

7. 更新数据

更新数据也非常简单，我们同样需要指定数据的 id 和内容，调用 update 方法即可，代码如下：

1
2
3
4
5
6
7
8
9
10

from elasticsearch import Elasticsearch

es = Elasticsearch()
data = {
    'title': '乘风破浪不负韶华，奋斗青春圆梦高考',
    'url': 'http://view.inews.qq.com/a/EDU2021041600732200',
    'date': '2021-07-05'
}
result = es.update(index='news', body=data, id=1)
print(result)

这里我们为数据增加了一个日期字段，然后调用了 update 方法，结果如下：

1

{'_index': 'news', '_type': '_doc', '_id': '1', '_version': 2, 'result': 'updated', '_shards': {'total': 2, 'successful': 1, 'failed': 0}, '_seq_no': 1, '_primary_term': 1}

可以看到，返回结果中 result 字段为 updated，即表示更新成功。另外，我们还注意到一个字段 _version，这代表更新后的版本号数，2 代表这是第二个版本。因为之前已经插入过一次数据，所以第一次插入的数据是版本 1，可以参见上例的运行结果，这次更新之后版本号就变成了 2，以后每更新一次，版本号都会加 1。

另外，更新操作利用 index 方法同样可以做到，其写法如下：

1

es.index(index='news', body=data, id=1)

可以看到，index 方法可以代替我们完成插入和更新数据这两个操作。如果数据不存在，就执行插入操作，如果已经存在，就执行更新操作，非常方便。

8. 删除数据

如果想删除一条数据，可以调用 delete 方法并指定需要删除的数据 id 即可。其写法如下：

1
2
3
4
5

from elasticsearch import Elasticsearch

es = Elasticsearch()
result = es.delete(index='news', id=1)
print(result)

运行结果如下：

1

{'_index': 'news', '_type': '_doc', '_id': '1', '_version': 2, 'result': 'deleted', '_shards': {'total': 2, 'successful': 1, 'failed': 0}, '_seq_no': 3, '_primary_term': 1}

可以看到，运行结果中 result 字段为 deleted，代表删除成功；_version 变成了 3，又增加了 1。

9. 查询数据

上面的几个操作都是非常简单的操作，普通的数据库如 MongoDB 都可以完成，看起来并没有什么了不起的，Elasticsearch 更特殊的地方在于其异常强大的检索功能。

对于中文来说，我们需要安装一个分词插件，这里使用的是 elasticsearch-analysis-ik，其 GitHub 链接为https://github.com/medcl/elasticsearch-analysis-ik。这里我们使用 Elasticsearch 的另一个命令行工具 elasticsearch-plugin 来安装，这里安装的版本是 7.13.2，请确保和 Elasticsearch 的版本对应起来，命令如下：

1

elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.13.2/elasticsearch-analysis-ik-7.13.2.zip

这里的版本号请替换成你的 Elasticsearch 版本号。

安装之后，我们需要重新启动 Elasticsearch，启动之后它会自动加载安装好的插件。

首先，我们重新新建一个索引并指定需要分词的字段，相应代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

from elasticsearch import Elasticsearch

es = Elasticsearch()
mapping = {
    'properties': {
        'title': {
            'type': 'text',
            'analyzer': 'ik_max_word',
            'search_analyzer': 'ik_max_word'
        }
    }
}
es.indices.delete(index='news', ignore=[400, 404])
es.indices.create(index='news', ignore=400)
result = es.indices.put_mapping(index='news', body=mapping)
print(result)

这里我们先将之前的索引删除了，然后新建了一个索引，接着更新了它的 mapping 信息。mapping 信息中指定了分词的字段，指定了字段的类型 type 为 text，分词器 analyzer 和搜索分词器 search_analyzer 为 ik_max_word，即使用我们刚才安装的中文分词插件。如果不指定的话，则使用默认的英文分词器。

接下来，我们插入几条新数据：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from elasticsearch import Elasticsearch

es = Elasticsearch()

datas = [
    {
        'title': '高考结局大不同',
        'url': 'https://k.sina.com.cn/article_7571064628_1c3454734001011lz9.html',
    },
    {
        'title': '进入职业大洗牌时代，“吃香”职业还吃香吗？',
        'url': 'https://new.qq.com/omn/20210828/20210828A025LK00.html',
    },
    {
        'title': '乘风破浪不负韶华，奋斗青春圆梦高考',
        'url': 'http://view.inews.qq.com/a/EDU2021041600732200',
    },
    {
        'title': '他，活出了我们理想的样子',
        'url': 'https://new.qq.com/omn/20210821/20210821A020ID00.html',
    }
]

for data in datas:
    es.index(index='news', body=data)

这里我们指定了 4 条数据，它们都带有 title 和url 字段，然后通过 index 方法将其插入 Elasticsearch 中，索引名称为 news。

接下来，我们根据关键词查询一下相关内容：

1
2

result = es.search(index='news')
print(result)

可以看到，这里查询出了插入的 4 条数据：

1

{'took': 11, 'timed_out': False, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 4, 'relation': 'eq'}, 'max_score': 1.0, 'hits': [{'_index': 'news', '_type': '_doc', '_id': 'jebpkHsBm-BAny-7hOYp', '_score': 1.0, '_source': {'title': '高考结局大不同', 'url': 'https://k.sina.com.cn/article_7571064628_1c3454734001011lz9.html'}}, {'_index': 'news', '_type': '_doc', '_id': 'jubpkHsBm-BAny-7hObz', '_score': 1.0, '_source': {'title': '进入职业大洗牌时代，“吃香”职业还吃香吗？', 'url': 'https://new.qq.com/omn/20210828/20210828A025LK00.html'}}, {'_index': 'news', '_type': '_doc', '_id': 'j-bpkHsBm-BAny-7heZN', '_score': 1.0, '_source': {'title': '乘风破浪不负韶华，奋斗青春圆梦高考', 'url': 'http://view.inews.qq.com/a/EDU2021041600732200'}}, {'_index': 'news', '_type': '_doc', '_id': 'kObpkHsBm-BAny-7hean', '_score': 1.0, '_source': {'title': '他，活出了我们理想的样子', 'url': 'https://new.qq.com/omn/20210821/20210821A020ID00.html'}}]}}

可以看到，返回结果会出现在 hits 字段里面，其中 total 字段标明了查询的结果条目数，max_score 代表了最大匹配分数。

另外，我们还可以进行全文检索，这才是体现 Elasticsearch 搜索引擎特性的地方：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

from elasticsearch import Elasticsearch
import json

dsl = {
    'query': {
        'match': {
            'title': '高考 圆梦'
        }
    }
}

es = Elasticsearch()
result = es.search(index='news', body=dsl)
print(result)

这里我们使用 Elasticsearch 支持的 DSL 语句来进行查询，使用 match 指定全文检索，检索的字段是 title，内容是“中国 领事馆”，搜索结果如下：

1

{'took': 6, 'timed_out': False, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 2, 'relation': 'eq'}, 'max_score': 1.7796917, 'hits': [{'_index': 'news', '_type': '_doc', '_id': 'j-bpkHsBm-BAny-7heZN', '_score': 1.7796917, '_source': {'title': '乘风破浪不负韶华，奋斗青春圆梦高考', 'url': 'http://view.inews.qq.com/a/EDU2021041600732200'}}, {'_index': 'news', '_type': '_doc', '_id': 'jebpkHsBm-BAny-7hOYp', '_score': 0.81085134, '_source': {'title': '高考结局大不同', 'url': 'https://k.sina.com.cn/article_7571064628_1c3454734001011lz9.html'}}]}}

这里我们看到匹配的结果有两条，第一条的分数为 1.7796917，第二条的分数为 0.81085134，这是因为第一条匹配的数据中含有“高考”和“圆梦”两个词，第二条匹配的数据中不包含“圆梦”，但是包含了“高考”这个词，所以也被检索出来了，但是分数比较低。

因此，可以看出，检索时会对对应的字段进行全文检索，结果还会按照检索关键词的相关性进行排序，这就是一个基本的搜索引擎雏形。

另外，Elasticsearch 还支持非常多的查询方式，这里就不再一一展开描述了，总之其功能非常强大，详情可以参考官方文档：https://www.elastic.co/guide/en/elasticsearch/reference/master/query-dsl.html。

10. 总结

以上便是对 Elasticsearch 的基本介绍以及使用 Python 操作 Elasticsearch 的基本用法，但这仅仅是 Elasticsearch 的基本功能，它还有更多强大的功能等待着我们去探索。

本节代码地址：https://github.com/Python3WebSpider/ElasticSearchTest。

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

我们知道爬虫是 IO 密集型任务，比如如果我们使用 requests 库来爬取某个站点的话，发出一个请求之后，程序必须要等待网站返回响应之后才能接着运行，而在等待响应的过程中，整个爬虫程序是一直在等待的，实际上没有做任何事情。对于这种情况，我们有没有优化方案呢？

当然有，下面我们就来了解一下异步爬虫的基本概念和实现。

要实现异步机制的爬虫，那自然和协程脱不了关系。

1. 案例引入

在介绍协程之前，我们先来看一个案例网站，链接地址为：https://httpbin.org/delay/5，如果我们访问这个链接，需要等待五秒之后才能得到结果，这是因为服务器强制等待了 5 秒的时间才返回响应。

平时我们浏览网页的时候，绝大部分网页响应速度还是很快的，如果我们写爬虫来爬取的话，发出 Request 到收到 Response 的时间不会很长，因此我们需要等待的时间并不多。

然而像上面这个网站，一次 Request 就需要 5 秒才能得到 Response，如果我们用 requests 写爬虫来爬取的话，那每次 requests 都要等待 5 秒才能拿到结果了。

我们来测试下，下面我们来用 requests 写一个遍历程序，直接遍历 100 次试试看，实现代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import requests
import logging
import time

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s: %(message)s')

TOTAL_NUMBER = 100
URL = 'https://httpbin.org/delay/5'

start_time = time.time()
for _ in range(1, TOTAL_NUMBER + 1):
    logging.info('scraping %s', URL)
    response = requests.get(URL)
end_time = time.time()
logging.info('total time %s seconds', end_time - start_time)

这里我们直接用循环的方式构造了 100 个 Request，使用的是 requests 单线程，在爬取之前和爬取之后记录了时间，最后输出爬取了 100 个页面消耗的时间。

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13

2020-08-03 01:01:36,781 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:01:43,410 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:01:50,029 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:01:56,702 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:02:03,345 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:02:09,958 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:02:16,500 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:02:23,143 - INFO: scraping https://httpbin.org/delay/5
...
2020-08-03 01:12:19,867 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:12:26,479 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:12:33,083 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:12:39,758 - INFO: total time 662.9764430522919 seconds

由于每个页面至少要等待 5 秒才能加载出来，因此 100 个页面至少要花费 500 秒的时间，加上网站本身负载的问题，总的爬取时间最终为 663 秒，大约 11 分钟。

这在实际情况下是很常见的，有些网站本身加载速度就比较慢，稍慢的可能 1~3 秒，更慢的说不定 10 秒以上。如果我们用 requests 单线程这么爬取的话，总的耗时是非常多的。此时如果我们开了多线程或多进程来爬取的话，其爬取速度确实会成倍提升，那是否有更好的解决方案呢？

本节就来了解一下使用协程来加速的方法，此种方法对于 IO 密集型任务非常有效。如将其应用到网络爬虫中，爬取效率甚至可以成百倍地提升。

2. 基础知识

在了解协程之前，我们首先了解一些基础概念，如阻塞和非阻塞、同步和异步、多进程和协程。

阻塞

阻塞状态指程序未得到所需计算资源时被挂起的状态。程序在等待某个操作完成期间，自身无法继续干别的事情，则称该程序在该操作上是阻塞的。

常见的阻塞形式有：网络 I/O 阻塞、磁盘 I/O 阻塞、用户输入阻塞等。阻塞是无处不在的，包括 CPU 切换上下文时，所有的进程都无法真正干事情，它们也会被阻塞。如果是多核 CPU，则正在执行上下文切换操作的核不可被利用。

非阻塞

程序在等待某操作的过程中，自身不被阻塞，可以继续运行干别的事情，则称该程序在该操作上是非阻塞的。

非阻塞并不是在任何程序级别、任何情况下都存在的。仅当程序封装的级别可以囊括独立的子程序单元时，它才可能存在非阻塞状态。

非阻塞的存在是因为阻塞存在，正因为某个操作阻塞导致的耗时与效率低下，我们才要把它变成非阻塞的。

同步

不同程序单元为了完成某个任务，在执行过程中需靠某种通信方式以协调一致，此时这些程序单元是同步执行的。

例如在购物系统中更新商品库存时，需要用“行锁”作为通信信号，让不同的更新请求强制排队顺序执行，那更新库存的操作是同步的。

简言之，同步意味着有序。

异步

为了完成某个任务，有时不同程序单元之间无须通信协调也能完成任务，此时不相关的程序单元之间可以是异步的。

例如，爬取下载网页。调度程序调用下载程序后，即可调度其他任务，而无须与该下载任务保持通信以协调行为。不同网页的下载、保存等操作都是无关的，也无须相互通知协调。这些异步操作的完成时刻并不确定。

简言之，异步意味着无序。

多进程

多进程就是利用 CPU 的多核优势，在同一时间并行执行多个任务，可以大大提高执行效率。

协程

协程，英文叫作 coroutine，又称微线程、纤程，它是一种用户态的轻量级线程。

协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此，协程能保留上一次调用时的状态，即所有局部状态的一个特定组合，每次过程重入时，就相当于进入上一次调用的状态。

协程本质上是个单进程，它相对于多进程来说，无须线程上下文切换的开销，无须原子操作锁定及同步的开销，编程模型也非常简单。

我们可以使用协程来实现异步操作，比如在网络爬虫场景下，我们发出一个请求之后，需要等待一定时间才能得到响应，但其实在这个等待过程中，程序可以干许多其他事情，等到响应得到之后才切换回来继续处理，这样可以充分利用 CPU 和其他资源，这就是协程的优势。

3. 协程的用法

接下来，让我们来了解一下协程的实现。从 Python 3.4 开始，Python 中加入了协程的概念，但这个版本的协程还是以生成器对象为基础，Python 3.5 则增加了 async/await，使得协程的实现更加方便。

Python 中使用协程最常用的库莫过于 asyncio，所以本节会以 asyncio 为基础来介绍协程的用法。

首先，我们需要了解下面几个概念：

• event_loop：事件循环，相当于一个无限循环，我们可以把一些函数注册到这个事件循环上，当满足条件发生的时候，就会调用对应的处理方法。
• coroutine：中文翻译叫协程，在 Python 中常指代协程对象类型，我们可以将协程对象注册到时间循环中，它会被事件循环调用。我们可以使用 async 关键字来定义一个方法，这个方法在调用时不会立即被执行，而是返回一个协程对象。
• task：任务，它是对协程对象的进一步封装，包含了任务的各个状态。
• future：代表将来执行或没有执行的任务的结果，实际上和 task 没有本质区别。

另外，我们还需要了解 async/await 关键字，它是从 Python 3.5 才出现的，专门用于定义协程。其中，async 定义一个协程，await 用来挂起阻塞方法的执行。

4. 准备工作

在本节开始之前，请确保安装的 Python 版本为 3.5 及以上，如果版本是 3.4 及以下，则下方的案例是不能运行的。

具体的安装方法可以参考：https://setup.scrape.center/python。

安装好合适的 Python 版本之后我们就可以开始本节的学习了。

5. 定义协程

首先，我们来定义一个协程，体验一下它和普通进程在实现上的不同之处，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12

import asyncio

async def execute(x):
    print('Number:', x)

coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')

loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine)
print('After calling loop')

运行结果如下：

1
2
3
4

Coroutine: <coroutine object execute at 0x1034cf830>
After calling execute
Number: 1
After calling loop

首先，我们引入了 asyncio 这个包，这样我们才可以使用 async 和 await，然后使用 async 定义了一个 execute 方法，该方法接收一个数字参数，执行之后会打印这个数字。

随后我们直接调用了这个方法，然而这个方法并没有执行，而是返回了一个 coroutine 协程对象。随后我们使用 get_event_loop 方法创建了一个事件循环 loop，并调用了 loop 对象的 run_until_complete 方法将协程注册到事件循环 loop 中，然后启动。最后，我们才看到 execute 方法打印了输出结果。

可见，async 定义的方法就会变成一个无法直接执行的 coroutine 对象，必须将其注册到事件循环中才可以执行。

前面我们还提到了 task，它是对 coroutine 对象的进一步封装，比 coroutine 对象多了运行状态，比如 running、finished 等，我们可以用这些状态来获取协程对象的执行情况。

在上面的例子中，当我们将 coroutine 对象传递给 run_until_complete 方法的时候，实际上它进行了一个操作，就是将 coroutine 封装成了 task 对象。我们也可以显式地进行声明，如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import asyncio

async def execute(x):
    print('Number:', x)
    return x

coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')

loop = asyncio.get_event_loop()
task = loop.create_task(coroutine)
print('Task:', task)
loop.run_until_complete(task)
print('Task:', task)
print('After calling loop')

运行结果如下：

1
2
3
4
5
6

Coroutine: <coroutine object execute at 0x10e0f7830>
After calling execute
Task: <Task pending coro=<execute() running at demo.py:4>>
Number: 1
Task: <Task finished coro=<execute() done, defined at demo.py:4> result=1>
After calling loop

这里我们定义了 loop 对象之后，接着调用了它的 create_task 方法将 coroutine 对象转化为 task 对象，随后我们打印输出一下，发现它是 pending 状态。接着，我们将 task 对象添加到事件循环中执行，随后打印输出 task 对象，发现它的状态变成了 finished，同时还可以看到其 result 变成了 1，也就是我们定义的 execute 方法的返回结果。

另外，定义 task 对象还有一种方式，就是直接通过 asyncio 的 ensure_future 方法，返回结果也是 task 对象，这样的话我们就可以不借助 loop 来定义。即使我们还没有声明 loop，也可以提前定义好 task 对象，写法如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import asyncio

async def execute(x):
    print('Number:', x)
    return x

coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')

task = asyncio.ensure_future(coroutine)
print('Task:', task)
loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:', task)
print('After calling loop')

运行结果如下：

1
2
3
4
5
6

Coroutine: <coroutine object execute at 0x10aa33830>
After calling execute
Task: <Task pending coro=<execute() running at demo.py:4>>
Number: 1
Task: <Task finished coro=<execute() done, defined at demo.py:4> result=1>
After calling loop

可以发现，其运行效果都是一样的。

6. 绑定回调

另外，我们也可以为某个 task 绑定一个回调方法。比如，我们来看下面的例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url)
    return status

def callback(task):
    print('Status:', task.result())

coroutine = request()
task = asyncio.ensure_future(coroutine)
task.add_done_callback(callback)
print('Task:', task)

loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:', task)

这里我们定义了一个 request 方法，请求了百度，获取其状态码，但是这个方法里面我们没有任何 print 语句。随后我们定义了一个 callback 方法，这个方法接收一个参数，是 task 对象，然后调用 print 方法打印了 task 对象的结果。这样我们就定义好了一个 coroutine 对象和一个回调方法。我们现在希望的效果是，当 coroutine 对象执行完毕之后，就去执行声明的 callback 方法。

那么它们两者怎样关联起来呢？很简单，只需要调用 add_done_callback 方法即可。我们将 callback 方法传递给封装好的 task 对象，这样当 task 执行完毕之后，就可以调用 callback 方法了。同时 task 对象还会作为参数传递给 callback 方法，调用 task 对象的 result 方法就可以获取返回结果了。

运行结果如下：

1
2
3

Task: <Task pending coro=<request() running at demo.py:5> cb=[callback() at demo.py:11]>
Status: <Response [200]>
Task: <Task finished coro=<request() done, defined at demo.py:5> result=<Response [200]>>

实际上不用回调方法，直接在 task 运行完毕之后，也可以直接调用 result 方法获取结果，如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url)
    return status

coroutine = request()
task = asyncio.ensure_future(coroutine)
print('Task:', task)

loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:', task)
print('Task Result:', task.result())

运行结果是一样的：

1
2
3

Task: <Task pending coro=<request() running at demo.py:4>>
Task: <Task finished coro=<request() done, defined at demo.py:4> result=<Response [200]>>
Task Result: <Response [200]>

7. 多任务协程

上面的例子我们只执行了一次请求，如果想执行多次请求，应该怎么办呢？我们可以定义一个 task 列表，然后使用 asyncio 的 wait 方法即可执行。看下面的例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url)
    return status

tasks = [asyncio.ensure_future(request()) for _ in range(5)]
print('Tasks:', tasks)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

for task in tasks:
    print('Task Result:', task.result())

这里我们使用一个 for 循环创建了 5 个 task，组成了一个列表，然后把这个列表首先传递给了 asyncio 的 wait 方法，再将其注册到时间循环中，就可以发起 5 个任务了。最后，我们再将任务的运行结果输出出来，具体如下：

1
2
3
4
5
6