用redis做简单的任务队列(二)


是用redis做任务队列时,要思考:

  • 用什么数据类型来做任务队列
  • 怎样才能防止重复爬取

上一篇文章已经决定使用list来做任务队列,但是去重问题没有得到解决。这里可以用set来解决思考二的问题,就是防止重复爬取的问题。

 

使用list当作未完成任务队列,存储还没有爬的url(或者是用户id,文章id等唯一标识)
使用set当作已完成任务队列,存储已经爬取的url
每次爬虫程序从list未完成任务队列获取任务的时候,都去set已完成任务队列里面验证一下,如果已完成队列里已经有了,就舍弃掉,如果没有,就开始爬取,并将这个url加入到已爬取的任务队列
这样做的方便之处在于:每当我往list未完成任务队列里加任务的时候,我不用考虑这个任务有没有爬过,这个任务是不是已经在未爬取任务队列了,我只需要往里加就行了,当爬虫去取的时候,让爬虫程序去做这个操作。

以下是具体代码 
算是一个生产消费把,master往队列里塞任务,parser使用get_html的返回值进行解析,然后入库。

协程爬取贴吧里发帖内容(redis做任务队列,mongo存储)

 1 import requests  2 from lxml import etree  3 import redis  4 import asyncio,aiohttp  5 
 6 import pymongo  7 conn = pymongo.MongoClient('localhost',27017)  8 
 9 db = conn.nicedb # 指定数据库名称,连接nicedb数据库,没有则自动创建
10 my_set = db.test_set # 使用test_set集合,没有则自动创建
11 # 以上两步都是延时操作,当往数据库插入第一条数据的时候,才会真正的创建数据库和集合
12 
13 # decode_responses=True,记得加这个参数,不加的话取出来的数据都是bytes类型的
14 r = redis.StrictRedis(host = '127.0.0.1', port = 6379, db = 2,decode_responses=True) 15 # pool = redis.ConnectionPool(host = '127.0.0.1', port = 6379, db = 2)
16 # r = redis.StrictRedis(connection_pool=pool,decode_responses=True)
17 
18 def master(page): 19     url = 'https://tieba.baidu.com/f?kw=美女&ie=utf-8&pn={}'.format(page*50) 20     base = 'https://tieba.baidu.com'
21     res = requests.get(url).text 22     html = etree.HTML(res) 23     half_urls = html.xpath("//div[@class='threadlist_title pull_left j_th_tit ']/a/@href") 24     full_urls = [base + i for i in half_urls] 25     for url in full_urls: 26         # 从url_list列表头部塞任务,也就是url
27         r.lpush('url_list',url) 28     #print(r.llen('url_list'))
29 
30 async def get_html(url): 31     async with asyncio.Semaphore(5):  # 限制并发数为5个
32  async with aiohttp.ClientSession() as session: 33  async with session.get(url) as html: 34                 # errors='ignore',不加这个参数的话,会报错,具体错误内容见下面图片
35                 response = await html.text(encoding='utf-8',errors='ignore') 36                 return response 37 async def parse(): 38     while True: 39         # 从redis的url_list列表取任务,从右边开始取
40         url = r.rpop('url_list') 41         if url == None: 42             break
43         # 判断这个任务是否已经做过了,也就是判断这个url在没在redis的history集合里
44         if r.sismember('history',url) == 1: 45             continue
46         response = await get_html(url) 47         html = etree.HTML(response) 48         content = html.xpath("//div[@class='left_section']/div[2]/div[1]//cc/div[1]/text()")[0].strip() 49         if content != '': 50             # 当内容不为空时,将内容存到mongo里
51             my_set.save({'content':content}) 52             #print(content)
53         # 将爬取过的任务放到redis的history集合里,也就是已完成任务队列
54         r.sadd('history', url) 55 t1 = time.time() 56 # 爬取前10页
57 for i in range(10): 58  master() 59 
60 # async的一些步骤
61 loop = asyncio.get_event_loop() 62 tasks = [parse() for _ in range(15)] 63 loop.run_until_complete(asyncio.wait(tasks)) 64 loop.close() 65 
66 t2 = time.time() 67 print(t2-t1) 68 # 最后用时:32.930299043655396
69 # 把mongo数据库换成mysql后,用时:43.06192493438721
70 
71 原文:https://blog.csdn.net/fiery_heart/article/details/82121237 

 


免责声明!

本站转载的文章为个人学习借鉴使用,本站对版权不负任何法律责任。如果侵犯了您的隐私权益,请联系本站邮箱yoyou2525@163.com删除。



 
粤ICP备18138465号  © 2018-2025 CODEPRJ.COM