Блог

Я использую RabbitMQ для своих пауков, каждый паук отправляет данные получателю.
Я запустил демон receiver.py as с помощью этой команды : daemon python /receiver.py

Когда я запускаю несколько экземпляров spider, у меня возникает ощущение, что receiver.py для очереди «истекло» больше одного экземпляра.

Что не так с моим кодом?

Отправитель работает следующим образом (Scrapy spider) :

 connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='expired')

def parse_items(self, response):
    for link in LxmlLinkExtractor(allow=(), deny=self.allowed_domains, canonicalize=False).extract_links(response):
         # self.logger.info('[{}] Added to the List'.format(domain))
         channel.basic_publish(exchange='', routing_key='expired', body=domain, )
         self.domains.append(domain)
  

Получатель делает это :

 class Threaded_worker(threading.Thread):

    def callback(self, ch, method, properties, domain):
        url = 'http://www.checkdomain.com/cgi-bin/checkdomain.pl?domain='   domain
        self.parse_checkdomain(url, domain)
        time.sleep(domain.count('.'))
        ch.basic_ack(delivery_tag=method.delivery_tag)

    def __init__(self):
        threading.Thread.__init__(self)
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()
        self.channel.queue_declare(queue='expired')
        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(self.callback, queue='expired')

    def run(self):
        logging.warning('Worker Start !')
        self.channel.start_consuming()


for _ in range(15):
    td = Threaded_worker()
    td.setDaemon(False)
    td.start()
  

Кстати, у меня есть еще один небольшой вопрос: если мой receiver.py не запускается, все данные по-прежнему сохраняются в очереди?