当包裹被送到一个地点后,需要分配到不同的地方才算整个过程的结束。那么把这个过程放到分布式爬虫中,Master节点负责了前期搬运和分类包裹的工作,剩下的任务小伙伴们都猜到了,spider_Worker节点就承担了最后的配送任务。我们也可以换种理解的方法,Worker是一个搬运工人的意思。那么我们今天就spider_Worker节点在python分布式爬虫中的使用给大家带来分享。
在将多线程版本改写成分布式的爬虫,主要用的可跨平台的multiprocessing.managers的BaseManager模块,这个模块的主要功能就是将task_queue和result_queue两个队列注册成函数暴露到网上去,Master节点监听端口,让Worker子节点去连接,不同主机之间就可以通过注册的函数来共享同步资源,Master节点主要负责发送任务和获取结果,Worker就获取任务队列的任务开始跑,并将获取的结果存储到数据库获取返回回来。Scrapy框架下的Spider类是什么
spider_Worker 节点主要调用spider()函数对任务进行处理,方法都类似,子节点每获取一个链接就传回Master, 另外需要注意的是Master文件只能运行一个,但Worker节点可以同时运行多个并行同步处理task任务队列。
spider_Master.py
#coding:utf-8 from multiprocessing.managers import BaseManager from Queue import Queue import time import argparse import MySQLdb import sys page = 2 word = 'inurl:login.action' output = 'test.txt' page = (page+1) * 10 host = '127.0.0.1' port = 500 urls = [] class Master(): def __init__(self): self.task_queue = Queue() #server需要先创建两个共享队列,worker端不需要 self.result_queue = Queue() def start(self): BaseManager.register('get_task_queue',callable=lambda:self.task_queue) #在网络上注册一个get_task_queue函数,即把两个队列暴露到网上,worker端不需要callable参数 BaseManager.register('get_result_queue',callable=lambda:self.result_queue) manager = BaseManager(address=(host,port),authkey='sir') manager.start() #master端为start,即开始监听端口,worker端为connect task = manager.get_task_queue() #master和worker都是从网络上获取task队列和result队列,不能在创建的两个队列 result = manager.get_result_queue() print 'put task' for i in range(0,page,10): target = 'https://www.baidu.com/s?wd=%s&pn=%s'%(word,i) print 'put task %s'%target task.put(target) print 'try get result' while True: try: url = result.get(True,5) #获取数据时需要超时长一些 print url urls.append(url) except: break manager.shutdown() if __name__ == '__main__': start = time.time() server = Master() server.start() print '共爬取数据%s条'%len(urls) print time.time()-start with open(output,'a') as f: for url in urls: f.write(url[1]+'\n') conn = MySQLdb.connect('localhost','root','root','Struct',charset='utf8') cursor = conn.cursor() for record in urls: sql = "insert into s045 values('%s','%s','%s')"%(record[0],record[1],str(record[2])) cursor.execute(sql) conn.commit() conn.close()
spider_Worker
#coding:utf-8 import re import Queue import time import requests from multiprocessing.managers import BaseManager from bs4 import BeautifulSoup as bs host = '127.0.0.1' port = 500 class Worder(): def __init__(self): self.headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64; rv:50.0) Gecko/20100101 Firefox/50.0'} def spider(self,target,result): urls = [] pn = int(target.split('=')[-1])/10 +1 # print pn # print target html = requests.get(target,headers=self.headers) soup = bs(html.text,"lxml") res = soup.find_all(name="a", attrs={'class':'c-showurl'}) for r in res: try: h = requests.get(r['href'],headers=self.headers,timeout=3) if h.status_code == 200: url = h.url # print url time.sleep(1) title = re.findall(r'<title>(.*?)</title>',h.content)[0] # print url,title title = title.decode('utf-8') print 'send spider url:',url result.put((pn,url,title)) else: continue except: continue # return urls def start(self): BaseManager.register('get_task_queue') BaseManager.register('get_result_queue') print 'Connect to server %s'%host m = BaseManager(address=(host,port),authkey='sir') m.connect() task = m.get_task_queue() result = m.get_result_queue() print 'try get queue' while True: try: target = task.get(True,1) print 'run pages %s'%target res = self.spider(target,result) # print res except: break if __name__ == '__main__': w = Worder() w.start()
看完本篇文章,相信大家对spider_Worker节点这个辛勤的搬运工有了新的认识,毕竟它承担了爬虫任务最后的配送环节,大家对最终的结果都是比较期待的。
神龙|纯净稳定代理IP免费测试>>>>>>>>天启|企业级代理IP免费测试>>>>>>>>IPIPGO|全球住宅代理IP免费测试