celery爬虫使用

简介

celery是一个基于分布式消息传输的异步任务队列&＃xff0c;它专注于实时处理&＃xff0c;同时也支持任务调度。

它由三部分组成&＃xff0c;消息中间件&＃xff0c;任务执行单元和任务执行结果存储组成。

官网 &＃xff1a;http://www.celeryproject.org/                 下载&＃xff1a;pip install celery

消息中间件&＃xff1a;Celery本身不提供消息服务&＃xff0c;但是可以方便的和第三方提供的消息中间件集成。包括&＃xff0c;RabbitMQ, Redis等等。
任务执行单元&＃xff1a;Worker是Celery提供的任务执行的单元&＃xff0c;worker并发的运行在分布式的系统节点中。
任务结果存储&＃xff1a;Task result store用来存储Worker执行的任务的结果&＃xff0c;Celery支持以不同方式存储任务的结果&＃xff0c;包括AMQP, redis等。Celery多用来执行异步任务&＃xff0c;将耗时的操作交由Celery去异步执行&＃xff0c;比如发送邮件、短信、消息推送、音视频处理等。还可以执行定时任务&＃xff0c;定时执行某件事情&＃xff0c;比如Redis中的数据每天凌晨两点保存至mysql数据库&＃xff0c;实现Redis的持久化。使用 Celery 之前请务必理解以下概念&＃xff1a;
a. Celery Beat: 任务调度器&＃xff0c;Beat 进程会读取配置文件的内容&＃xff0c;周期性的将配置中到期需要执行的任务发送给任务队列
b. Celery Worker: 执行任务的消费者&＃xff0c;通常会在多台服务器运行多个消费者来提高运行效率。
c. Broker: 消息代理&＃xff0c;也是任务队列本身&＃xff08;通常是消息队列或者数据库&＃xff09;&＃xff0c;通常称为消息中间件&＃xff0c;接收任务生产者发送过来的任务消息&＃xff0c;存进队列再按序分发给任务消费方。
d. Producer: 任务生产者&＃xff0c;调用 Celery API 的函数或者装饰器而产生任务并交给任务队列处理的都是任务生产者。

项目结构如下&＃xff1a;

其中&＃xff0c;app_test.py为主程序&＃xff0c;其代码如下&＃xff1a;

from celery import Celeryapp &＃61; Celery(&＃39;proj&＃39;, include&＃61;[&＃39;proj.tasks&＃39;])
app.config_from_object(&＃39;proj.celeryconfig&＃39;)if __name__ &＃61;&＃61; &＃39;__main__&＃39;:app.start()

tasks.py为任务函数&＃xff0c;代码如下&＃xff1a;

import re
import requests
from celery import group
from proj.app_test import app&＃64;app.task(trail&＃61;True)
# 并行调用任务
def get_content(urls):return group(C.s(url) for url in urls)()&＃64;app.task(trail&＃61;True)
def C(url):return parser.delay(url)&＃64;app.task(trail&＃61;True)
# 获取每个网页的name和description
def parser(url):req &＃61; requests.get(url)html &＃61; req.texttry:name &＃61; re.findall(r&＃39;(.&＃43;?)&＃39;, html)[0]desc &＃61; re.findall(r&＃39;(.&＃43;?)&＃39;, html)[0]if name is not None and desc is not None:return name, descexcept Exception as err:return &＃39;&＃39;, &＃39;&＃39;

celeryconfig.py为celery的配置文件&＃xff0c;代码如下&＃xff1a;

BROKER_URL &＃61; &＃39;redis://localhost&＃39; # 使用Redis作为消息代理CELERY_RESULT_BACKEND &＃61; &＃39;redis://localhost:6379/0&＃39; # 把任务结果存在了RedisCELERY_TASK_SERIALIZER &＃61; &＃39;msgpack&＃39; # 任务序列化和反序列化使用msgpack方案CELERY_RESULT_SERIALIZER &＃61; &＃39;json&＃39; # 读取任务结果一般性能要求不高&＃xff0c;所以使用了可读性更好的JSONCELERY_TASK_RESULT_EXPIRES &＃61; 60 * 60 * 24 # 任务过期时间CELERY_ACCEPT_CONTENT &＃61; [&＃39;json&＃39;, &＃39;msgpack&＃39;] # 指定接受的内容类型

最后是我们的爬虫文件&＃xff0c;scrapy.py&＃xff0c;代码如下&＃xff1a;

import time
import requests
from bs4 import BeautifulSoup
from proj.tasks import get_contentt1 &＃61; time.time()
url &＃61; "http://www.wikidata.org/w/index.php?title&＃61;Special:WhatLinksHere/Q5&limit&＃61;500&from&＃61;0"headers &＃61; {&＃39;User-Agent&＃39;: &＃39;Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, \like Gecko) Chrome/67.0.3396.87 Safari/537.36&＃39;}req &＃61; requests.get(url, headers&＃61;headers)
soup &＃61; BeautifulSoup(req.text, "lxml")
human_list &＃61; soup.find(id&＃61;&＃39;mw-whatlinkshere-list&＃39;)(&＃39;li&＃39;)urls &＃61; []
for human in human_list:url &＃61; human.find(&＃39;a&＃39;)[&＃39;href&＃39;]urls.append(&＃39;https://www.wikidata.org&＃39;&＃43;url)
#print(urls)# 调用get_content函数&＃xff0c;并获取爬虫结果
result &＃61; get_content.delay(urls)
res &＃61; [v for v in result.collect()]for r in res:if isinstance(r[1], list) and isinstance(r[1][0], str):print(r[1])t2 &＃61; time.time() # 结束时间
print(&＃39;耗时&＃xff1a;%s&＃39; % (t2 - t1))

在后台启动redis&＃xff0c;并切换至proj项目所在目录&＃xff0c;运行命令&＃xff1a;

celery -A proj.app_test worker -l info

输出结果如下(只显示最后几行的输出)&＃xff1a;

......
[&＃39;Antoine de Saint-Exupery&＃39;, &＃39;French writer and aviator&＃39;]
[&＃39;&＃39;, &＃39;&＃39;]
[&＃39;Sir John Barrow, 1st Baronet&＃39;, &＃39;English statesman&＃39;]
[&＃39;Amy Johnson&＃39;, &＃39;pioneering English aviator&＃39;]
[&＃39;Mike Oldfield&＃39;, &＃39;English musician, multi-instrumentalist&＃39;]
[&＃39;Willoughby Newton&＃39;, &＃39;politician from Virginia, USA&＃39;]
[&＃39;Mack Wilberg&＃39;, &＃39;American conductor&＃39;]
耗时&＃xff1a;80.05160284042358

在rdm中查看数据&＃xff0c;如下&＃xff1a;

Celery进阶—分布式爬虫

新建文件crawlertask.py&＃xff0c;用于执行数据抓取任务&＃xff0c;代码如下。

#coding:utf-8import requests
from bs4 import BeautifulSoup
from celery import Celery,platformsapp &＃61; Celery(&＃39;tasks&＃39;,broker&＃61;&＃39;redis://localhost:6379/0&＃39;)
app.conf.CELERY_RESULT_BACKEND&＃61;&＃39;redis://localhost:6379/0&＃39;platforms.C_FORCE_ROOT&＃61;Truedef format_str(str):return str.replace("\n","").replace(" ","").replace("\t","")&＃64;app.task
def get_urls_in_pages(from_page_num,to_page_num):urls&＃61;[]search_word&＃61;&＃39;计算机&＃39;url_part_1&＃61;&＃39;http://www.phei.com.cn/module/goods/&＃39;\&＃39;searchkey.jsp?Page&＃61;&＃39;url_part_2&＃61;&＃39;&Page&＃61;2&searchKey&＃61;&＃39;for i in range(from_page_num,to_page_num&＃43;1):urls.append(url_part_1&＃43;str(i)&＃43;url_part_2&＃43;search_word)all_href_list&＃61;[]for url in urls:resp&＃61;requests.get(url)bs&＃61;BeautifulSoup(resp.text)a_list&＃61;bs.find_all(&＃39;a&＃39;)needed_list&＃61;[]for a in a_list:if &＃39;href&＃39; in a.attrs:href_val&＃61;a[&＃39;href&＃39;]title&＃61;a.textif &＃39;bookid&＃39; in href_val and &＃39;shopcar0.jsp&＃39; not in href_val and title!&＃61;&＃39;&＃39;:if [title,href_val] not in needed_list:needed_list.append([format_str(title),format_str(href_val)])all_href_list&＃43;&＃61;needed_listall_href_file &＃61; open(str(from_page_num)&＃43;&＃39;_&＃39;&＃43;str(to_page_num)&＃43;&＃39;_&＃39;&＃43;&＃39;all_hrefs.txt&＃39;,&＃39;w&＃39;)for href in all_href_list:all_href_file.write(&＃39;\t&＃39;.join(href)&＃43;&＃39;\n&＃39;)all_href_file.close()return len(all_href_list)


【部署服务器方法】

将以上脚本部署到两台云端服务器 , 并且在云端开启redis服务&＃xff0c;然后执行&＃xff1a;

celery worker -A crawlertask -l info -c 10

【部署从机方法】

将上面的脚本部署到两台主机A和B&＃xff0c;然后各自运行下面的命令&＃xff1a;

celery -A crawl_douban worker -l info

在本机新建文件task_dist.py用于异步分发任务&＃xff0c;代码如下&＃xff1a;

from celery import Celery
from threading import Thread
import timeredis_ips&＃61;{0:&＃39;redis://101.200.163.195:6379/0&＃39;,1:&＃39;redis://112.124.28.41:6379/0&＃39;,2:&＃39;redis://112.124.28.41:6379/0&＃39;,3:&＃39;redis://101.200.163.195:6379/0&＃39;,
}def send_task_and_get_results(ind,from_page,to_page):app&＃61;Celery(&＃39;crawlertask&＃39;,broker&＃61;redis_ips[ind])app.conf.CELERY_RESULT_BACKEND&＃61;redis_ips[ind]result&＃61;app.send_task(&＃39;crawlertask.get_urls_in_pages&＃39;,args&＃61;(from_page,to_page))print(redis_ips[ind],result.get())if __name__&＃61;&＃61;&＃39;__main__&＃39;:t1&＃61;time.time()page_ranges_lst&＃61;[(1,10),(11,20),(21,30),(31,40),]th_lst &＃61; []for ind, page_range in enumerate(page_ranges_lst):th &＃61; Thread(target&＃61;send_task_and_get_results,args&＃61;(ind,page_range[0], page_range[1]))th_lst.append(th)for th in th_lst:th.start()for th in th_lst:th.join()t2 &＃61; time.time()print("用时&＃xff1a;", t2 - t1)

举例&＃xff1a;

以爬douban小说为例 首先启动Redis&＃xff0c;新建文件crawl_douban.py

import requests
import time
import redis
from celery import Celery
from bs4 import BeautifulSoup
from configparser import ConfigParsercp&＃61;ConfigParser()
cp.read(&＃39;config&＃39;)#获取配置信息
db_host&＃61;cp.get(section&＃61;&＃39;redis&＃39;,option&＃61;&＃39;db_host&＃39;)
db_port&＃61;cp.getint(&＃39;redis&＃39;,&＃39;db_port&＃39;)
db_pwd&＃61;cp&＃39;redis&＃39;#redis连接
pool &＃61; redis.ConnectionPool(host&＃61;db_host, port&＃61;db_port, db&＃61;15, password&＃61;db_pwd)
r &＃61; redis.StrictRedis(connection_pool&＃61;pool)
set_name&＃61;&＃39;crawl:douban&＃39;app &＃61; Celery(&＃39;crawl&＃39;, include&＃61;[&＃39;task&＃39;], broker&＃61;&＃39;redis://:{}&＃64;{}:{}/12&＃39;.format(db_pwd,db_host,db_port), backend&＃61;&＃39;redis://:{}&＃64;{}:{}/13&＃39;.format(db_pwd,db_host,db_port))#官方推荐使用json作为消息序列化方式
app.conf.update(CELERY_TIMEZONE&＃61;&＃39;Asia/Shanghai&＃39;,CELERY_ENABLE_UTC&＃61;True,CELERY_ACCEPT_CONTENT&＃61;[&＃39;json&＃39;],CELERY_TASK_SERIALIZER&＃61;&＃39;json&＃39;,CELERY_RESULT_SERIALIZER&＃61;&＃39;json&＃39;,
)headers&＃61;{&＃39;User-Agent&＃39;:&＃39;&＃39;,}&＃64;app.task
def crawl(url):res&＃61;requests.get(url,headers&＃61;headers)#延迟2秒time.sleep(2)soup&＃61;BeautifulSoup(res.text,&＃39;lxml&＃39;)items&＃61;soup.select(&＃39;.subject-list .subject-item .info h2 a&＃39;)titles&＃61;[item[&＃39;title&＃39;] for item in items]#将小说的title存入redis数据库r.sadd(set_name,(url,titles,time.time()))print(titles)return (url,titles)

将上面的脚本部署到两台主机A和B&＃xff0c;然后各自运行下面的命令&＃xff1a;

celery -A crawl_douban worker -l info

在本机C新建文件task_dispatcher.py用于异步分发任务&＃xff0c;代码如下&＃xff1a;

from crawl_douban import app
from crawl_douban import crawldef manage_crawl(urls):for url in urls:app.send_task(&＃39;crawl_douban.crawl&＃39;, args&＃61;(url,))#上句也可以写成 crawl.apply_async(args&＃61;(url,)) 或 crawl.delay(url)if __name__ &＃61;&＃61; &＃39;__main__&＃39;:start_url &＃61; &＃39;https://book.douban.com/tag/小说&＃39;#爬去10页&＃xff0c;每页20本书url_list &＃61; [&＃39;{}?start&＃61;{}&type&＃61;T&＃39;.format(start_url, page * 20) for page in range(10)]manage_crawl(url_list)

运行task_dispatcher.py&＃xff0c;跑完用时2.8s

celery worker -A tasks --loglevel&＃61;info --concurrency&＃61;5# 参数”-A”指定了Celery实例的位置
# 参数”loglevel”指定了日志等级&＃xff0c;也可以不加&＃xff0c;默认为warning。
# 参数”concurrency”指定最大并发数&＃xff0c;默认为CPU核数。