python抓取异步

1. 从零开始学python-使用Selenium抓取动态网页数据

AJAX（Asynchronouse JavaScript And XML：异步JavaScript和XML）通过在后台与服务器进行少量数据交换，Ajax 可以使网页实现异步更新，这意味着可以在不重新加载整个网页的情况下，对网页的某部分进行局部更新。传统的网页（不使用Ajax）如果需要更新内容，必须重载整个网页页面。

因为传统的网页在传输数据格式方面，使用的是 XML 语法，因此叫做 AJAX ，其实现在数据交互基本上都是使用 JSON 。使用AJAX加载的数据，即使使用了JS将数据渲染到了浏览器中，在 右键->查看网页源代码 还是不能看到通过ajax加载的数据，只能看到使用这个url加载的html代码。

法1：直接分析ajax调用的接口。然后通过代码请求这个接口。

法2：使用Selenium+chromedriver模拟浏览器行为获取数据。

Selenium 相当于是一个机器人。可以模拟人类在浏览器上的一些行为，自动处理浏览器上的一些行为，比如点击，填充数据，删除cookie等。 chromedriver 是一个驱动 Chrome 浏览器的驱动程序，使用他才可以驱动浏览器。当然针对不同的浏览器有不同的driver。以下列出了不同浏览器及其对应的driver：

现在以一个简单的获取网络首页的例子来讲下 Selenium 和 chromedriver 如何快速入门：

参考：Selenium的使用

直接直接分析ajax调用的接口爬取

selenium结合lxml爬取

2. Python 异步任务队列Celery 使用

在 Python 中定义 Celery 的时候，我们要引入 Broker，中文翻译过来就是“中间人”的意思。在工头(生产者)提出任务的时候，把所有的任务放到 Broker 里面，在 Broker 的另外一头，一群码农(消费者)等着取出一个个任务准备着手做。这种模式注定了整个系统会是个开环系统，工头对于码农们把任务做的怎样是不知情的。所以我们要引入 Backend 来保存每次任务的结果。这个 Backend 也是存储任务的信息用的，只不过这里存的是那些任务的返回结果。我们可以选择只让错误执行的任务返回结果到 Backend，这样我们取回结果，便可以知道有多少任务执行失败了。

其实现架构如下图所示：

可以看到，Celery 主要包含以下几个模块：

celery可以通过pip自动安装。

broker 可选择使用RabbitMQ/redis，backend可选择使用RabbitMQ/redis/MongoDB。RabbitMQ/redis/mongoDB的安装请参考对应的官方文档。

------------------------------rabbitmq相关----------------------------------------------------------

官网安装方法： http://www.rabbitmq.com/install-windows.html

启动管理插件：sbin/rabbitmq-plugins enable rabbitmq_management 启动rabbitmq：sbin/rabbitmq-server -detached

rabbitmq已经启动，可以打开页面来看看 地址： http://localhost:15672/#/

用户名密码都是guest 。进入可以看到具体页面。 关于rabbitmq的配置，网上很多 自己去搜以下就ok了。

------------------------------rabbitmq相关--------------------------------------------------------

项目结构如下：

使用前，需要三个方面：celery配置，celery实例，需执行的任务函数，如下：

Celery 的配置比较多，可以在 官方配置文档： http://docs.celeryproject.org/en/latest/userguide/configuration.html 查询每个配置项的含义。

当然，要保证上述异步任务and下述定时任务都能正常执行，就需要先启动celery worker，启动命令行如下:

需 启动beat ，执行定时任务时, Celery会通过celery beat进程来完成。Celery beat会保持运行, 一旦到了某一定时任务需要执行时, Celery beat便将其加入到queue中. 不像worker进程, Celery beat只需要一个即可。而且为了避免有重复的任务被发送出去，所以Celery beat仅能有一个。

命令行启动：

如果你想将celery worker/beat要放到后台运行，推荐可以扔给supervisor。

supervisor.conf如下：

3. python爬取大量数据(百万级)

当用python爬取大量网页获取想要的数据时，最重要的问题是爬虫中断问题，python这种脚本语言，一中断

进程就会退出，怎么在中断后继续上次爬取的任务就至关重要了。这里就重点剖析这个中断问题。

第一个问题: 简单点的用动态代理池就能解决，在爬取大量数据的时候，为了速度不受影响，建议使用一些缓

存的中间件将有效的代理 ip 缓存起来，并定时更新。这里推荐 github 这个仓库

https://github.com/jhao104/proxy_pool ， 它会做ip有效性验证并将 ip 放入 redis ，不过实现过于复杂

了，还用到了 db ，个人觉得最好自己修改一下。困难点的就是它会使用别的请求来进行判断当前的ip是否

是爬虫，当我们过于聚焦我们的爬虫请求而忽略了其他的请求时，可能就会被服务器判定为爬虫，进而这个ip

会被列入黑名单，而且你换了ip一样也会卡死在这里。这种方式呢，简单点就用 selenium + chrome 一个一个

去爬，不过速度太慢了。还是自己去分析吧，也不会过复杂的。

第二个问题： 网络连接超时是大概率会遇到的问题，有可能是在爬取的时候本地网络波动，也有可能是爬

取的服务端对ip做了限制，在爬取到了一定量级的时候做一些延迟的操作，使得一些通用的 http 库超时

（ urllib ）。不过如果是服务端动的手脚一般延迟不会太高，我们只需要人为的设置一个高一点的

timeout 即可（30 秒），最好在爬取开始的时候就对我们要用的爬取库进行一层封装，通用起来才好改

动。

第三个问题: 在解析大量静态页面的时候，有些静态页面的解析规则不一样，所以我们就必须得做好断点

续爬的准备了（ PS : 如果简单的忽略错误可能会导致大量数据的丢失，这就不明智了）。那么在调试的过

程中断点续爬有个解决方案，就是生产者和消费者分离，生产者就是产生待爬 url 的爬虫，消费者就是爬取

最终数据的爬虫。最终解析数据就是消费者爬虫了。他们通过消息中间件连接，生产者往消息中间件发送待

爬取的目标信息，消费者从里面取就行了，还间接的实现了个分布式爬取功能。由于现在的消费中间件都有

ack 机制，一个消费者爬取链接失败会导致消息消费失败，进而分配给其他消费者消费。所以消息丢失的

概率极低。不过这里还有个 tips ， 消费者的消费超时时间不能太长，会导致消息释放不及时。还有要开启

消息中间价的数据持久化功能，不然消息产生过多而消费不及时会撑爆机器内存。那样就得不偿失了。

第四个问题： 这种情况只能 try except catch 住了，不好解决，如果单独分析的话会耗费点时间。但在

大部分数据 （99%） 都正常的情况下就这条不正常抛弃就行了。主要有了第三个问题的解决方案再出现这

种偶尔中断的问就方便多了。

希望能帮到各位。

4. python用什么可以异步爬去数据

可以使用线程方式 在线程中爬取数据 主线程原来做控制 也可以使用多进程方式 用进程池方式

5. 如何用Python爬取数据

方法/步骤

• 在做爬取数据之前，你需要下载安装两个东西，一个是urllib,另外一个是python-docx。

  

• 7

  这个爬下来的是源代码，如果还需要筛选的话需要自己去添加各种正则表达式。

6. python异步有哪些方式

yield相当于return，他将相应的值返回给调用next()或者send()的调用者，从而交出了CPU使用权，而当调用者再次调用next()或者send()的时候，又会返回到yield中断的地方，如果send有参数，还会将参数返回给yield赋值的变量,如果没有就和next（）一样赋值为None。但是这里会遇到一个问题，就是嵌套使用generator时外层的generator需要写大量代码，看如下示例：
注意以下代码均在Python3.6上运行调试

#!/usr/bin/env python# encoding:utf-8def inner_generator():
i = 0
while True:
i = yield i if i > 10: raise StopIterationdef outer_generator():
print("do something before yield")
from_inner = 0
from_outer = 1
g = inner_generator()
g.send(None) while 1: try:
from_inner = g.send(from_outer)
from_outer = yield from_inner except StopIteration: breakdef main():
g = outer_generator()
g.send(None)
i = 0
while 1: try:
i = g.send(i + 1)
print(i) except StopIteration: breakif __name__ == '__main__':
main()041

为了简化，在Python3.3中引入了yield from

yield from

使用yield from有两个好处，

1、可以将main中send的参数一直返回给最里层的generator，
2、同时我们也不需要再使用while循环和send (), next()来进行迭代。

我们可以将上边的代码修改如下：

def inner_generator():
i = 0
while True:
i = yield i if i > 10: raise StopIterationdef outer_generator():
print("do something before coroutine start") yield from inner_generator()def main():
g = outer_generator()
g.send(None)
i = 0
while 1: try:
i = g.send(i + 1)
print(i) except StopIteration: breakif __name__ == '__main__':
main()

执行结果如下：

do something before coroutine start123456789101234567891011

这里inner_generator()中执行的代码片段我们实际就可以认为是协程，所以总的来说逻辑图如下：

我们都知道Python由于GIL(Global Interpreter Lock)原因，其线程效率并不高，并且在*nix系统中，创建线程的开销并不比进程小，因此在并发操作时，多线程的效率还是受到了很大制约的。所以后来人们发现通过yield来中断代码片段的执行，同时交出了cpu的使用权，于是协程的概念产生了。在Python3.4正式引入了协程的概念，代码示例如下：

import asyncio# Borrowed from http://curio.readthedocs.org/en/latest/[email protected] countdown(number, n):
while n > 0:
print('T-minus', n, '({})'.format(number)) yield from asyncio.sleep(1)
n -= 1loop = asyncio.get_event_loop()
tasks = [
asyncio.ensure_future(countdown("A", 2)),
asyncio.ensure_future(countdown("B", 3))]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()12345678910111213141516

示例显示了在Python3.4引入两个重要概念协程和事件循环，
通过修饰符@asyncio.coroutine定义了一个协程，而通过event loop来执行tasks中所有的协程任务。之后在Python3.5引入了新的async & await语法，从而有了原生协程的概念。

async & await

在Python3.5中，引入了aync&await 语法结构，通过”aync def”可以定义一个协程代码片段，作用类似于Python3.4中的@asyncio.coroutine修饰符，而await则相当于”yield from”。

先来看一段代码，这个是我刚开始使用async&await语法时，写的一段小程序。

#!/usr/bin/env python# encoding:utf-8import asyncioimport requestsimport time


async def wait_download(url):
response = await requets.get(url)
print("get {} response complete.".format(url))


async def main():
start = time.time()
await asyncio.wait([
wait_download("http://www.163.com"),
wait_download("http://www.mi.com"),
wait_download("http://www.google.com")])
end = time.time()
print("Complete in {} seconds".format(end - start))


loop = asyncio.get_event_loop()
loop.run_until_complete(main())

这里会收到这样的报错：

Task exception was never retrieved
future: <Task finished coro=<wait_download() done, defined at asynctest.py:9> exception=TypeError("object Response can't be used in 'await' expression",)>
Traceback (most recent call last):
File "asynctest.py", line 10, in wait_download
data = await requests.get(url)
TypeError: object Response can't be used in 'await' expression123456

这是由于requests.get()函数返回的Response对象不能用于await表达式，可是如果不能用于await，还怎么样来实现异步呢？
原来Python的await表达式是类似于”yield from”的东西，但是await会去做参数检查，它要求await表达式中的对象必须是awaitable的，那啥是awaitable呢？ awaitable对象必须满足如下条件中其中之一：

1、A native coroutine object returned from a native coroutine function .

原生协程对象

2、A generator-based coroutine object returned from a function decorated with types.coroutine() .

types.coroutine()修饰的基于生成器的协程对象，注意不是Python3.4中asyncio.coroutine

3、An object with an await method returning an iterator.

实现了await method，并在其中返回了iterator的对象

根据这些条件定义，我们可以修改代码如下：

#!/usr/bin/env python# encoding:utf-8import asyncioimport requestsimport time


async def download(url): # 通过async def定义的函数是原生的协程对象
response = requests.get(url)
print(response.text)


async def wait_download(url):
await download(url) # 这里download(url)就是一个原生的协程对象
print("get {} data complete.".format(url))


async def main():
start = time.time()
await asyncio.wait([
wait_download("http://www.163.com"),
wait_download("http://www.mi.com"),
wait_download("http://www.google.com")])
end = time.time()
print("Complete in {} seconds".format(end - start))


loop = asyncio.get_event_loop()
loop.run_until_complete(main())27282930

好了现在一个真正的实现了异步编程的小程序终于诞生了。
而目前更牛逼的异步是使用uvloop或者pyuv，这两个最新的Python库都是libuv实现的，可以提供更加高效的event loop。

uvloop和pyuv

pyuv实现了Python2.x和3.x，但是该项目在github上已经许久没有更新了，不知道是否还有人在维护。
uvloop只实现了3.x, 但是该项目在github上始终活跃。

它们的使用也非常简单，以uvloop为例，只需要添加以下代码就可以了

import asyncioimport uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())123

7. python2.7怎么实现异步

改进之前
之前，我的查询步骤很简单，就是：
前端提交查询请求 --> 建立数据库连接 --> 新建游标 --> 执行命令 --> 接受结果 --> 关闭游标、连接
这几大步骤的顺序执行。
这里面当然问题很大：
建立数据库连接实际上就是新建一个套接字。这是进程间通信的几种方法里，开销最大的了。
在“执行命令”和“接受结果”两个步骤中，线程在阻塞在数据库内部的运行过程中，数据库连接和游标都处于闲置状态。
这样一来，每一次查询都要顺序的新建数据库连接，都要阻塞在数据库返回结果的过程中。当前端提交大量查询请求时，查询效率肯定是很低的。
第一次改进
之前的模块里，问题最大的就是第一步——建立数据库连接套接字了。如果能够一次性建立连接，之后查询能够反复服用这个连接就好了。
所以，首先应该把数据库查询模块作为一个单独的守护进程去执行，而前端app作为主进程响应用户的点击操作。那么两条进程怎么传递消息呢？翻了几天Python文档，终于构思出来：用队列queue作为生产者（web前端）向消费者（数据库后端）传递任务的渠道。生产者，会与SQL命令一起，同时传递一个管道pipe的连接对象，作为任务完成后，回传结果的渠道。确保，任务的接收方与发送方保持一致。
作为第二个问题的解决方法，可以使用线程池来并发获取任务队列中的task，然后执行命令并回传结果。
第二次改进
第一次改进的效果还是很明显的，不用任何测试手段。直接点击页面链接，可以很直观地感觉到反应速度有很明显的加快。
但是对于第二个问题，使用线程池还是有些欠妥当。因为，CPython解释器存在GIL问题，所有线程实际上都在一个解释器进程里调度。线程稍微开多一点，解释器进程就会频繁的切换线程，而线程切换的开销也不小。线程多一点，甚至会出现“抖动”问题（也就是刚刚唤醒一个线程，就进入挂起状态，刚刚换到栈帧或内存的上下文，又被换回内存或者磁盘），效率大大降低。也就是说，线程池的并发量很有限。
试过了多进程、多线程，只能在单个线程里做文章了。
Python中的asyncio库
Python里有大量的协程库可以实现单线程内的并发操作，比如Twisted、Gevent等等。Python官方在3.5版本里提供了asyncio库同样可以实现协程并发。asyncio库大大降低了Python中协程的实现难度，就像定义普通函数那样就可以了，只是要在def前面多加一个async关键词。async def函数中，需要阻塞在其他async def函数的位置前面可以加上await关键词。
import asyncio
async def wait():
await asyncio.sleep(2)
async def execute(task):
process_task(task)
await wait()
continue_job()
async def函数的执行稍微麻烦点。需要首先获取一个loop对象，然后由这个对象代为执行async def函数。
loop = asyncio.get_event_loop()
loop.run_until_complete(execute(task))
loop.close()
loop在执行execute(task)函数时，如果遇到await关键字，就会暂时挂起当前协程，转而去执行其他阻塞在await关键词的协程，从而实现协程并发。
不过需要注意的是，run_until_complete()函数本身是一个阻塞函数。也就是说，当前线程会等候一个run_until_complete()函数执行完毕之后，才会继续执行下一部函数。所以下面这段代码并不能并发执行。
for task in task_list:
loop.run_until_complete(task)
对与这个问题，asyncio库也有相应的解决方案：gather函数。
loop = asyncio.get_event_loop()
tasks = [asyncio.ensure_future(execute(task))
for task in task_list]
loop.run_until_complete(asyncio.gather(*tasks))
loop.close()
当然了，async def函数的执行并不只有这两种解决方案，还有call_soon与run_forever的配合执行等等，更多内容还请参考官方文档。
Python下的I/O多路复用
协程，实际上，也存在上下文切换，只不过开销很轻微。而I/O多路复用则完全不存在这个问题。
目前，Linux上比较火的I/O多路复用API要算epoll了。Tornado，就是通过调用C语言封装的epoll库，成功解决了C10K问题（当然还有Pypy的功劳）。
在Linux里查文档，可以看到epoll只有三类函数，调用起来比较方便易懂。
创建epoll对象，并返回其对应的文件描述符（file descriptor）。
int epoll_create(int size);
int epoll_create1(int flags);
控制监听事件。第一个参数epfd就对应于前面命令创建的epoll对象的文件描述符；第二个参数表示该命令要执行的动作：监听事件的新增、修改或者删除；第三个参数，是要监听的文件对应的描述符；第四个，代表要监听的事件。
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
等候。这是一个阻塞函数，调用者会等候内核通知所注册的事件被触发。
int epoll_wait(int epfd, struct epoll_event *events,
int maxevents, int timeout);
int epoll_pwait(int epfd, struct epoll_event *events,
int maxevents, int timeout,
const sigset_t *sigmask);
在Python的select库里：
select.epoll()对应于第一类创建函数；
epoll.register()，epoll.unregister()，epoll.modify()均是对控制函数epoll_ctl的封装；
epoll.poll()则是对等候函数epoll_wait的封装。
Python里epoll相关API的最大问题应该是在epoll.poll()。相比于其所封装的epoll_wait，用户无法手动指定要等候的事件，也就是后者的第二个参数struct epoll_event *events。没法实现精确控制。因此只能使用替代方案：select.select()函数。
根据Python官方文档，select.select(rlist, wlist, xlist[, timeout])是对Unix系统中select函数的直接调用，与C语言API的传参很接近。前三个参数都是列表，其中的元素都是要注册到内核的文件描述符。如果想用自定义类，就要确保实现了fileno()方法。
其分别对应于：
rlist: 等候直到可读
wlist: 等候直到可写
xlist: 等候直到异常。这个异常的定义，要查看系统文档。
select.select()，类似于epoll.poll()，先注册文件和事件，然后保持等候内核通知，是阻塞函数。
实际应用
Psycopg2库支持对异步和协程，但和一般情况下的用法略有区别。普通数据库连接支持不同线程中的不同游标并发查询；而异步连接则不支持不同游标的同时查询。所以异步连接的不同游标之间必须使用I/O复用方法来协调调度。
所以，我的大致实现思路是这样的：首先并发执行大量协程，从任务队列中提取任务，再向连接池请求连接，创建游标，然后执行命令，并返回结果。在获取游标和接受查询结果之前，均要阻塞等候内核通知连接可用。
其中，连接池返回连接时，会根据引用连接的协程数量，返回负载最轻的连接。这也是自己定义AsyncConnectionPool类的目的。
我的代码位于：bottle-blog/dbservice.py
存在问题
当然了，这个流程目前还一些问题。
首先就是每次轮询拿到任务之后，都会走这么一个流程。
获取连接 --> 新建游标 --> 执行任务 --> 关闭游标 --> 取消连接引用
本来，最好的情况应该是：在轮询之前，就建好游标；在轮询时，直接等候内核通知，执行相应任务。这样可以减少轮询时的任务量。但是如果协程提前对应好连接，那就不能保证在获取任务时，保持各连接负载均衡了。
所以这一块，还有工作要做。
还有就是epoll没能用上，有些遗憾。
以后打算写点C语言的内容，或者用Python/C API，或者用Ctypes包装共享库，来实现epoll的调用。
最后，请允许我吐槽一下Python的epoll相关文档：简直太弱了！！！必须看源码才能弄清楚功能。