pyspider + RabbitMQ 使用记 - 下

老白的释然

首先我们需要安装 RabbitMQ，然后通过服务启动它，默认为注册到本机的5672端口。我们的爬虫和数据库写入脚本都需要连接到 RabbitMQ，一边往队列中写入数据，另一边从队列中取出数据，然后插入到数据。

Python 中使用 RabbitMQ 可以通过调用 pika 这个库，安装过程见官方文档，对于 RabbitMQ 本身也有中文教程。

本项目用到的模型是一对一的，用 pika 写很容易，代码如下：

1. import pika
   
2. # 导入库
   
3. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
   
4. # 设置一个新连接，连接到本地的 RabbitMQ 服务端。
   
5. channel = connection.channel()
   
6. # 注册到 books 队列
   
7. channel.queue_declare(queue='books')
   
8. channel.basic_publish(exchange='', routing_key='books', body='Whats up')
   
9. # 发送消息 body
   
10. connection.close()
    
11. #

复制代码

在 basic_publish 这个函数中，我们设置 exchange 为空，而 routing-key 为 books，此时 basic_publish 会默认把我们的 body 信息根据 routing-key 的内容发送到 books 的队列中。

这里 exchange 其实是一个信息中转站，如下图，P 为我们要发送的信息，X 就是信息中转站，我们通过 exchange 字段来设置我们的目标中转站，然后由 exchange 来决定我们的信息要往哪里走。

而 routing-key 的设置也很有讲究，可以参考教程中 Routing 一节。

到此，我们已经写好生产者了，接下来我们看消费者。

1. import pika
   
2. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
   
3. channel = connection.channel()
   
4. channel.queue_declare(queue='books')
   
5. 
   
6. 
   
7. def callback(ch, method, properties, body):
   
8.     print body
   
9. 
   
10. channel.basic_consume(callback, queue='books', no_ack=True)
    
11. # 注册回调函数，当有消息取出时，程序调用 callback 函数，其中 body 就是取出的消息。
    
12. channel.start_consuming()

复制代码

这里我们在回调函数中可以直接打印 body 到控制台。

最后贴一下代码，爬虫端：

1. #!/usr/bin/env python
   
2. # -*- encoding: utf-8 -*-
   
3. 
   
4. from pyspider.libs.base_handler import *
   
5. import pika
   
6. 
   
7. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
   
8. channel = connection.channel()
   
9. channel.queue_declare(queue='books')
   
10. 
    
11. 
    
12. class Handler(BaseHandler):
    
13.     crawl_config = {}
    
14. 
    
15.     def on_start(self):
    
16.         self.crawl('http://scrapy.org/', callback=self.index_page)
    
17.    
    
18.     def on_result(self, result):
    
19.         # 重写 on_result 函数
    
20.         if not result:
    
21.             return
    
22.         assert self.task, "on_result can't outside a callback."
    
23.         result['callback'] = self.task['process']['callback']
    
24.         if self.is_debugger():
    
25.             pprint(result)
    
26.             channel.basic_publish(exchange='', routing_key='books', body=result)
    
27.         if self.__env__.get('result_queue'):
    
28.             channel.basic_publish(exchange='', routing_key='books', body=result)
    
29.             self.__env__['result_queue'].put((self.task, result))
    
30.         
    
31.     @config(priority=5)
    
32.     def index_page(self, response):
    
33.         url_list = []
    
34.         for each in response.doc('a[href^="http"]').items():
    
35.             url_list.append(each.attr.href)
    
36.         return url_list

复制代码

消费者端：

1. import os
   
2. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'spider.settings')
   
3. import django
   
4. django.setup()
   
5. from django.core.exceptions import ObjectDoesNotExist
   
6. from importer.models import Books
   
7. 
   
8. import pika
   
9. connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
   
10. channel = connection.channel()
    
11. channel.queue_declare(queue='books')
    
12. 
    
13. 
    
14. def callback(ch, method, properties, body):
    
15.     new_book = Books()
    
16.     new_book.url = body
    
17.     new_book.save()
    
18.     print body + ' saved!'
    
19. 
    
20. channel.basic_consume(callback, queue='books', no_ack=True)
    
21. channel.start_consuming()

复制代码

本例中调用了 django 的 BOOKS 的数据库模型，在队列中取出消息后，存入 BOOKS 表中，字段为url。

Miss_love

支持分享