Gevent中爬虫与多线程详解

谁践踏了优雅 2022-07-13 10:37 251阅读 0赞

**gevent****实现爬虫并与普通多线程比较**

1.爬虫目的与思路简介；

2.gevent实现爬虫；

3.多线程实现爬虫；

4.对比。

**爬虫目的与设计**

1.从一个主页爬取相关链接；

2.比如麦子学院主站；

3.http://www.maiziedu.com；

4.从此网站爬取内容，分析过滤有效的不重复的网址（不包含图片，视频等资源地址）；

5.但是为了控制结束条件，可以设置爬虫深入和总数目，并可以适当控制并发度。

**gevent实现爬虫**

主要利用greenlet来执行具体爬虫操作，pool来控制并发度。

**代码实例：**

\#!/usr/bin/env python

\#-\*- coding: utf-8 -\*-

import re

import urllib

import urlparse

from pyquery import PyQuery

class HtmlAnalyzer(object):

@staticmethod

def detectCharSet(html):

pq = PyQuery(html)

metas = pq('head')('meta')

for meta in metas:

for key in meta.keys():

if key == "charset":

charset = meta.get('charset')

return charset

elif key == "content":

try:

content = meta.get('content')

p = re.match(r".+charset=(.\*)\\W\*", content)

charset = p.group(1)

return charset

except:

continue

@staticmethod

def extractLinks(html, baseurl, charset):

def \_extract(url, attr):

link = url.attrib\[attr\]

link = link.strip("/ ").strip('\\\\"')

if link is None:

raise

link = urlparse.urljoin(baseurl, link)

link = urlparse.urldefrag(link)\[0\]

try:

link = urllib.quote(link, ':?=+&\#/@')

except (UnicodeDecodeError, KeyError):

try:

link = urllib.quote(link.encode(charset), ':?=+&\#/@')

except:

pass

return link

def \_isValidLink(url):

try:

return all(\[UrlFilter.checkScheme(url),

UrlFilter.checkInvalidChar(url),

UrlFilter.checkInvalidExtention(url)

\])

except:

return False

pq = PyQuery(html)

allLinks = \[\]

for url in pq('a'):

try:

link = \_extract(url, 'href')

except:

continue

if \_isValidLink(link):

allLinks.append(link)

for url in pq('form'):

try:

link = \_extract(url, 'action')

except:

continue

if \_isValidLink(link):

allLinks.append(link)

return allLinks

class UrlFilter(object):

invalid\_chars = \{'\\'': None,

'\\"': None,

'\\\\': None,

' ': None,

'\\n': None,

'\\r': None,

'+': None

invalid\_extention = \{

'jpg': None,

'gif': None,

'bmp': None,

'jpeg': None,

'png': None,

'swf': None,

'mp3': None,

'wma': None,

'wmv': None,

'wav': None,

'mid': None,

'ape': None,

'mpg': None,

'mpeg': None,

'rm': None,

'rmvb': None,

'avi': None,

'mkv': None,

'zip': None,

'rar': None,

'gz': None,

'iso': None,

'jar': None,

'doc': None,

'docx': None,

'ppt': None,

'pptx': None,

'chm': None,

'pdf': None,

'exe': None,

'msi': None,

@staticmethod

def checkScheme(url):

scheme, netloc, path, pm, q, f = urlparse.urlparse(url)

return scheme in ('http', 'https')

@classmethod

def checkInvalidChar(cls, url):

exist\_invalid\_char = False

for c in url:

if c in cls.invalid\_chars:

exist\_invalid\_char = True

break

return (not exist\_invalid\_char)

@classmethod

def checkInvalidExtention(cls, url):

dotpos = url.rfind('.') + 1

typestr = url\[dotpos:\].lower()

return (typestr not in cls.invalid\_extention)

Geventspider.py

import gevent

from gevent import (

monkey,

queue,

pool

)

import requests

from threading import Timer

from utils import HtmlAnalyzer

class Strategy(object):

def \_\_init\_\_(self, max\_depth, max\_count, concurrency=5):

self.max\_depth = max\_depth

self.max\_count = max\_count

self.concurrency = concurrency

self.timeout = 60

self.time = 12\*60

class UrlObject(object):

def \_\_init\_\_(self, url, depth):

self.url = url.strip('/')

self.depth = depth

class GeventSpider(object):

def \_\_init\_\_(self, max\_depth, max\_count, root\_url):

monkey.patch\_all()

self.strategy = Strategy(max\_depth=max\_depth, max\_count=max\_count)

self.queue = queue.Queue()

self.pool = pool.Pool(self.strategy.concurrency)

self.url\_set = set()

obj = UrlObject(root\_url, 0)

self.put(obj)

self.handle\_num = 0

def put(self, obj):

hash\_val = hash(obj.url)

if hash\_val not in self.url\_set:

self.url\_set.add(hash\_val)

self.queue.put(obj)

def stop(self):

self.timer.cancel()

self.pool.join()

def \_run\_loop(self):

while self.timer.isAlive():

for geeenlet in list(self.pool):

if geeenlet.dead:

self.pool.discard(geeenlet)

try:

url = self.queue.get()

except queue.Empty:

continue

greentlet = Handler(url, self)

self.pool.start(greentlet)

self.handle\_num = self.handle\_num + 1

if self.strategy.max\_count <= self.handle\_num:

print 'handlernum %d is full so stop' %self.handle\_num

self.stop()

def run(self):

self.timer = Timer(self.strategy.time, self.stop)

self.timer.start()

self.\_run\_loop()

class Handler(gevent.Greenlet):

def \_\_init\_\_(self, urlobj, spider):

print 'begin greenlet with url ', urlobj.url

gevent.Greenlet.\_\_init\_\_(self)

self.urlobj = urlobj

self.spider = spider

self.charset = "utf-8"

def open(self, url):

strategy = self.spider.strategy

try:

resp = requests.get(url, timeout = strategy.timeout)

except:

return

if resp.status\_code != requests.codes.ok:

return

charset = HtmlAnalyzer.detectCharSet(resp.text)

if charset is not None:

self.charset = charset

resp.encoding = charset

return resp.text

def \_run(self):

try :

html = self.open(self.urlobj.url)

except Exception,why:

return

depth = self.urlobj.depth + 1

if depth > self.spider.strategy.max\_depth:

return

for link in self.feed(html):

if hash(link) in self.spider.url\_set:

continue

url = UrlObject(link, depth)

self.spider.put(url)

def feed(self, html):

return HtmlAnalyzer.extractLinks(html, self.urlobj.url, self.charset)

class MySpider(object):

def \_\_init\_\_(self, max\_depth, max\_count, root\_url):

self.spider = GeventSpider(max\_depth=max\_depth,

max\_count=max\_count,

root\_url=root\_url)

def run(self):

self.spider.run()

test = MySpider(max\_depth=5, max\_count=100, root\_url="http://www.maiziedu.com")

test.run()

Threadspider.py

import requests

from threading import Timer, Lock, Thread

from utils import HtmlAnalyzer

from Queue import Queue

import time

class Strategy(object):

def \_\_init\_\_(self, max\_depth, max\_count, concurrency=5):

self.max\_depth = max\_depth

self.max\_count = max\_count

self.concurrency = concurrency

self.timeout = 60

self.time = 12\*60

class UrlObject(object):

def \_\_init\_\_(self, url, depth):

self.url = url.strip('/')

self.depth = depth

class ThreadSpider(object):

def \_\_init\_\_(self, max\_depth, max\_count, root\_url):

self.strategy = Strategy(max\_depth, max\_count)

self.queue = Queue()

self.url\_set = set()

self.handler\_num = 0

self.lock = Lock()

self.thread\_lock = Lock()

self.thread\_pool = \{\}

self.thread\_id = 0

self.is\_stop = False

self.thread\_num = 0

self.currency\_limit = False

self.last\_data = None

obj = UrlObject(root\_url, 0)

self.put(obj)

def put(self, obj):

hash\_val = hash(obj.url)

self.lock.acquire()

res = hash\_val in self.url\_set

self.lock.release()

if res:

return

self.url\_set.add(hash\_val)

self.queue.put(obj)

def \_run\_loop(self):

while True:

if self.is\_stop:

time.sleep(1)

continue

if self.currency\_limit:

time.sleep(1)

self.thread\_lock.acquire()

self.thread\_num = len(self.thread\_pool)

if self.thread\_num == self.strategy.concurrency:

self.thread\_lock.release()

continue

else:

self.currency\_limit = False

self.thread\_lock.release()

else:

try:

url = self.queue.get()

except:

continue

self.thread\_id = self.thread\_id+1

thd = Handler(url, self, self.thread\_id)

self.thread\_lock.acquire()

self.thread\_pool\[self.thread\_id\] = thd

if len(self.thread\_pool) == self.strategy.concurrency:

self.currency\_limit = True

self.thread\_lock.release()

self.thread\_num = self.thread\_num+1

print "add thread ", self.thread\_id

thd.start()

self.handler\_num = self.handler\_num+1

if self.strategy.max\_count <= self.handler\_num:

print "handler num %d is full so stop " % self.handler\_num

self.is\_stop = True

def remove\_thread(self, thd\_id):

self.thread\_lock.acquire()

if thd\_id in self.thread\_pool:

del self.thread\_pool\[thd\_id\]

print "del threadid ", thd\_id

self.thread\_lock.release()

def run(self):

self.\_run\_loop()

class Handler(Thread):

def \_\_init\_\_(self, urlobj, spider, thd\_id):

Thread.\_\_init\_\_(self)

print "begin thread %d with url %s" %(thd\_id, urlobj.url)

self.urlobj= urlobj

self.spider = spider

self.thread\_id = thd\_id

self.charset = "utf-8"

def run(self):

try :

html = self.open(self.urlobj.url)

except Exception,why:

return

depth = self.urlobj.depth + 1

if depth > self.spider.strategy.max\_depth:

return

for link in self.feed(html):

if hash(link) in self.spider.url\_set:

continue

url = UrlObject(link, depth)

self.spider.put(url)

self.spider.remove\_thread(self.thread\_id)

def open(self, url):

strategy = self.spider.strategy

try:

resp = requests.get(url, timeout=strategy.timeout)

except:

return

if resp.status\_code != requests.codes.ok:

return

charset = HtmlAnalyzer.detectCharSet(resp.text)

if charset is not None:

self.charset = charset

resp.encoding = charset

return resp.text

def feed(self, html):

return HtmlAnalyzer.extractLinks(html, self.urlobj.url, self.charset)

class MySpider(object):

def \_\_init\_\_(self, max\_depth, max\_count, root\_url):

self.spider = ThreadSpider(max\_depth=max\_depth,

max\_count=max\_count,

root\_url=root\_url)

def run(self):

self.spider.run()

test = MySpider(max\_depth=5, max\_count=100, root\_url="http://www.maiziedu.com")

test.run()

**thread实现爬虫**

1.在一个线程内执行爬虫动作；

2.操作全局数据时注意线程安全。

**对比**

1.线程操作复杂，线程安全安全问题非常繁琐，容易出问题，而去不易掌握，甚至出现死锁，很难调试；

2.gevent模式相对简单，控制并发、调试都容易。

原文链接：[http://www.maiziedu.com/wiki/frame/creeper/][http_www.maiziedu.com_wiki_frame_creeper]

[http_www.maiziedu.com_wiki_frame_creeper]: http://www.maiziedu.com/wiki/frame/creeper/

Gevent中爬虫与多线程详解

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关【Linux】多线程详解（中）

相关 Python多线程爬虫——数据分析项目实现详解

相关 Java多线程中ReentrantLock与Condition详解

相关 Java并发与多线程-详解线程池

相关 python多线程队列爬虫流程图_Python+多线程+队列爬虫

相关 Gevent中爬虫与多线程详解

相关多线程详解

相关多线程爬虫

相关 Python3爬虫与多线程

相关多线程爬虫