2018-02-03

爬虫

如何让爬虫更快

本文章属于爬虫入门到精通系统教程第十一讲

在前面的教程中，我们已经学会了如何抓取一个网页，可是，当我需要抓取的数据足够多的时候，应该如何让我抓取的速度更快呢？

最简单的方法就是使用多进程.

什么是多线程

多线程（英语：multithreading），是指从软件或者硬件上实现多个线程并发执行的技术。具有多线程能力的计算机因有硬件支持而能够在同一时间执行多于一个线程，进而提升整体处理性能。

使用多线程的好处

使用线程可以把占据时间长的程序中的任务放到后台去处理
用户界面可以更加吸引人，这样比如用户点击了一个按钮去触发某些事件的处理，可以弹出一个进度条来显示处理的进度
程序的运行速度可能加快
在一些等待的任务实现上如用户输入、文件读写和网络收发数据等，线程就比较有用了。在这种情况下可以释放一些珍贵的资源如内存占用等等。
多线程技术在IOS软件开发中也有举足轻重的位置。

在python中使用多线程

安装方法：

pip install futures

注意：由于本系列教程是基于python2的，所以需要安装，如果你使用的是python3 ，在标准库里已经有这个模块了。直接用就行

concurrent的介绍

concurrent.futures 模块为异步执行可调用的对象提供了一个高级的接口。

异步执行可以通过线程来实现，使用 ThreadPoolExecutor 模块，或者使用 ProcessPoolExecutor 模块通过分离进程来实现。两种实现都有同样的接口，他们都是通过抽象类 Executor 来定义的。

Executor 对象

class concurrent.futures.Executor

这是一个抽象类，用来提供方法去支持异步地执行调用，它不应该被直接调用，而是应该通过具体的子类来使用。

submit(fn, *args, **kwargs)
可调用对象的调度器，fn参数将会以fn(*args, **kwargs)的形式来调用，同时返回一个 Future 对象代表了可调用对象的执行情况。

1
2
3

with ThreadPoolExecutor(max_workers=1) as executor:
  future = executor.submit(pow, 323, 1235)
  print(future.result())

map(func, *iterables, timeout=None, chunksize=1)

和map(func, *iterables)函数的作用基本相同，除了func是被异步执行的，而且几个对于func调用可能是同时执行的。这个函数返回的迭代器调用__next__()方法的时候，如果在timeout秒内结果不可用，那么迭代器将会从原始调用的函数向Executor.map()抛出一个concurrent.futures.TimeoutError的异常。timeout既能是一个整数，也能是一个浮点数。如果timeout没有指定的话或者等于 None 的话，那么等待时间就没有限制。如果调用函数抛出了一个异常，那么当迭代器取到这个函数的时候，异常将会被抛出。
当使用ProcessPoolExecutor的时候，这个方法将iterables切成许多块，然后将这些内容作为分离的任务提交到进程池中。每个块的大概的尺寸能够通过chunksize(大于0的正整数)的参数来指定。当iterables非常大的时候，和chunksize默认等于1相比，将chunksize设置为一个很大的值，将会显著地提升性能。在使用ThreadPoolExecutor的情况下，chunksize的大小没有影响。

Python 3.5新增功能：添加了chunksize参数

shutdown(wait=True)

告诉执行器，当当前阻塞的 futures 执行完了以后，它应该释放所有它使用的资源。在shutdown函数之后再来调用Executor.submit()和Executor.map()将会抛出RuntimeError

如果wait等于 True 的话，这个方法不会立即返回，而直到所有阻塞的 futures 都返回，而且和这个执行器所有相关的资源都被释放以后，这个函数才会返回。如果wait设置为 False ，那么这个方法会立刻返回，而和这个执行器所有相关的资源只有等到所有阻塞的 futures 都执行完以后才会被释放。而无论wait参数的值是什么，整个 Python 程序都会等到所有阻塞的 futures 执行完毕以后才会退出。

通过with语句，可以避免明确地来调用这个方法，它在执行完以后将会自动关闭Executor。(调用 Executor.shutdown() 时wait会被设置为True，这将会等待所有 future 执行完毕)

import shutil
with ThreadPoolExecutor(max_workers=4) as e:
  e.submit(shutil.copy, 'src1.txt', 'dest1.txt')
  e.submit(shutil.copy, 'src2.txt', 'dest2.txt')
  e.submit(shutil.copy, 'src3.txt', 'dest3.txt')
  e.submit(shutil.copy, 'src4.txt', 'dest4.txt')

ThreadPoolExecutor(重点)

ThreadPoolExecutor是Executor的子类，使用一个线程池去异步地执行调用。

class concurrent.futures.ThreadPoolExecutor(max_workers=None)

一个Executor的子类，使用线程池中最多max_workers个线程去异步地执行回调。
Python 3.5中的改变：如果max_workers参数为None或者没有给定，那么它将会被默认设置成为机器的CPU核数乘5。这里假设ThreadPoolExecutor经常被用来执行IO密集型的工作而不是CPU密集型的工作，工作者的个数应该比ProcessPoolExecutor的工作者的个数要多。

ThreadPoolExecutor 例子

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
'http://www.cnn.com/',
'http://europe.wsj.com/',
'http://www.bbc.co.uk/',
'http://some-made-up-domain.com/']

# 获取一个单页，同时报告URL和内容
def load_url(url, timeout):
  with urllib.request.urlopen(url, timeout=timeout) as conn:
  return conn.read()

# 我们通过with语句来确保线程能够被及时地清理，
# 这边max_workers=5，表示最多同时有5个线程去执行
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
  # 字典生成器,使用的方法是`executor.submit()`
  future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
  '''
  字典生成器用for循环实现的话，如下
  future_to_url = {}
  for url in URLS:
  future = executor.submit(load_url, url, 60)
  future_to_url[future] = url
  '''
  for future in concurrent.futures.as_completed(future_to_url):
  url = future_to_url[future]
  # 这边的future就是 通过`Executor.submit()`函数来创建的。
  #有以下常用方法方法
  # future.result(),返回由相关回调产生的结果，在本案列中，返回函数`load_url`的结果
  # future.exception() 返回由相关回调抛出的异常，如果没有异常则返回`None`
  # 更多future对象介绍请看下文
  if future.exception() is not None:
    print('%r generated an exception: %s' % (url,future.exception()))
  else:
    print('%r page is %d bytes' % (url, len(future.result())))

Future 对象

Future 类封装了一个可调用对象的异步执行过程，Future 对象是通过Executor.submit()函数来创建的。

class concurrent.futures.Future

封装了一个可调用对象的异步执行过程。Future 实例是通过Executor.submit()方法来创建的，而且不应该被直接创建，除非用来测试。

cancel()
尝试去取消相关回调，如果这个回调正在被执行，而且不能被取消，那么这个方法将会返回False，否则这个方法将会取消相应的回调并且返回True

cancelled()
如果相关回调被成功取消了，那么这个方法将会返回True

running()
如果相关回调当前正在被执行而且无法取消，那么将会返回True

done()
如果相关的回调被成功地取消或者已经运行完毕那么将返回True

result(timeout=None)
返回由相关回调产生的结果。如果这个回调还没有被完成那么这个方法将会等待timeout秒。如果这个回调在timeout秒内还没有返回，一个concurrent.futures.TimeoutError的异常将会被抛出。timeout能够被设置成一个整数或者一个浮点数。如果timeout没有被设置或者其值为None，那么等待时间将没有限制。

如果这个 future 在完成之前被取消了，那么将会抛出一个CancelledError的异常。
如果相关的回调抛出了一个异常，那么这个方法也会相应地抛出这个异常。

exception(timeout=None)
返回由相关回调抛出的异常。如果相关回调还没有被完成那么这个方法将会等待timeout秒。如果相关回调在timeout秒内还没有被完成，那么将会抛出一个concurrent.futures.TimeoutError的异常。timeout能够被设置成一个整数或者一个浮点数。如果timeout没有被设置或者其值为None，那么等待时间将没有限制。

如果这个 future 在完成之前被取消了，那么将会抛出一个CancelledError的异常。
如果相关回调被完成了且没有抛出异常，None将会被返回。

add_done_callback(fn)
将可调用对象fn连接到这个 future 上，fn将会在 future 被取消或者结束运行时被调用，而且仅有相关 future 这一个参数。
添加的可调用对象将会以它们被添加的顺序来调用，而且总是在添加它们的那个进程的所属的线程中调用(译者注，可以参考这段代码)。如果相关调用fn抛出了一个Exception子类的异常，它将会被记录和忽略。如果相关调用fn抛出了一个BaseException子类的异常，那么行为是未定义的。
如果相关的 future 已经被完成了或者取消了，fn将会被立刻调用。

实例

# coding: utf-8
import cPickle as pickle
import requests
from concurrent.futures import ThreadPoolExecutor
from concurrent import futures
from pymongo import MongoClient
client = MongoClient()
db = client.loc
collection = db.mobai1

def load_url(url, params, timeout, headers=None):
    return requests.get(url, params=params, timeout=timeout, headers=headers).json()


def getloc():
    allloc = []
    """利用高德地图api获取上海所有的小区坐标
    http://lbs.amap.com/api/webservice/guide/api/search/#text
    """
    with ThreadPoolExecutor(max_workers=5) as executor:
        url = 'http://restapi.amap.com/v3/place/text'
        param = {
            'key': '22d6f93f929728c10ed86258653ae14a',
            'keywords': u'小区',
            'city': '021',
            'citylimit': 'true',
            'output': 'json',
            'page': '',
        }
        future_to_url = {executor.submit(load_url, url, merge_dicts(param, {'page': i}), 60): url for i in range(1, 46)}
        for future in futures.as_completed(future_to_url):
            if future.exception() is not None:
                print future.exception()
            elif future.done():
                data = future.result()['pois']
                allloc.extend([x['location'] for x in data])
        with open('allloc1.pk', 'wb') as f:
            pickle.dump(allloc, f, True)


def merge_dicts(*dict_args):
    '''
   可以接收1个或多个字典参数
    '''
    result = {}
    for dictionary in dict_args:
        result.update(dictionary)
    return result


def mobai(loc):
    allmobai = []
    with ThreadPoolExecutor(max_workers=5) as executor:
        url = 'https://mwx.mobike.com/mobike-api/rent/nearbyBikesInfo.do'
        headers = {
            'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 10_3_1 like Mac OS X) AppleWebKit/603.1.30 (KHTML, like Gecko) Mobile/14E304 MicroMessenger/6.5.7 NetType/WIFI Language/zh_CN',
            'Content-Type': 'application/x-www-form-urlencoded',
            'Referer': 'https://servicewechat.com/wx80f809371ae33eda/23/page-frame.html',
        }
        data = {
            'longitude': '',
            'latitude': '',
            'citycode': '021',
        }
        future_to_url = {
            executor.submit(load_url, url, merge_dicts(data, {'longitude': i.split(',')[0]}, {'latitude': i.split(',')[1]}),
                            60,headers): url for i in loc}
        for future in futures.as_completed(future_to_url):
            if future.exception() is not None:
                print future.exception()
            elif future.done():
                data = future.result()['object']
                allmobai.extend(data)
                # 存入mongodb
                result = collection.insert_many(data)

总结

完本编文章，你应该学会“如何使用多线程抓取网页”
最后所有代码都在https://github.com/jin10086/pachong/blob/master/mobai.py

本文作者：高金
本文地址： https://igaojin.me/2018/02/03/如何让爬虫更快/
版权声明：转载请注明出处！

高金的博客