原文章 ：https://www.cnblogs.com/chnmig/p/11987948.html#1974317082
使用RabbitMQ作为任务队列#
使用RabbitMQ作为任务队列的轮子很少,基本都已停止更新(17年),这里推荐一个国人的修改版,最近才开始发布,但是经本人实测已经可以正常使用
项目地址(GitHub)scrapy-rabbitmq-scheduler
因是国人写的,所以README文件写的通俗易懂.
安装#
pip install scrapy-rabbitmq-scheduler
集成至Scrapy#
在settings.py最后加入
<pre style="margin: 10px 0px; padding: 0px; white-space: pre !important; overflow-wrap: break-word; position: relative !important;">
Copy
`# 指定项目的调度器
SCHEDULER = "scrapy_rabbitmq_scheduler.scheduler.SaaS"
指定rabbitmq的连接DSN
amqp_url="amqp://username:password@ip:port/"
RABBITMQ_CONNECTION_PARAMETERS = 'amqp://admin:pwd@x.x.x.x:5672/'
指定重试的http状态码(重新加回队列重试)
如果结果的状态码位该list其中一个则会重试
SCHEDULER_REQUEUE_ON_STATUS = [500]
指定下载器中间件, 确认任务是否成功
DOWNLOADER_MIDDLEWARES = {
'scrapy_rabbitmq_scheduler.middleware.RabbitMQMiddleware': 999
}
指定item处理方式, item会加入到rabbitmq中
ITEM_PIPELINES = {
'scrapy_rabbitmq_scheduler.pipelines.RabbitmqPipeline': 300,
}` </pre>
爬虫编写#
这里与Scrapy原来的方式稍有不同
构造发送请求/接收RabbitMQ数据的方法名为 _make_request
我们必须重构该方法才可正常运行使用爬虫
该方法起到每次从队列中拿取数据后的解析数据并进行请求的作用
通常我们存放在队列中的一个数据为一个JSON/msgpack格式,里面包含了要请求的URl/该条数据所属ID等多个信息
必须要注意的是如果遇到跳转或你在setting中设置了返回状态码为xxx重新爬取,那么Scrapy会将需要重新爬取的url存放至你的队列中,此时队列中有两种格式的数据
- 上游生产者发送的符合你的通信协议的数据(JSON/msgpack)
- Scrapy自己加入的需要重新爬取的数据
所以我们要对这两种数据加以区分
爬虫示例
image
item是对返回数据序列化的,item
image
运行爬虫后,scrapy获取到的数据会转JSON传送至res_status_ok中(你自定义的)
添加个Work在队列另一端接收入库或其他操作即可
需要注意的点#
Scrapy自己插入数据到分发队列#
当时被这个问题卡了一小时,网上是没有解决方法的,他生成的数据也是不能使用常规方法进行解码的,这常常令人一头雾水
通道设置#
该组件默认RabbitMQ持久化为True,因此请注意建立通道的时候将设置对齐否则会出现因为设置错误导致无法连接的问题
