RabbitMQ怎么在python分布式爬蟲中構建

這篇文章主要介紹RabbitMQ怎么在python分布式爬蟲中構建，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！

RabbitMQ是比較靠譜的消息中間件，得益于它的確認機制，當一條消息消費后如果設置確定模式，那么確認后才會繼續消費，如果不確定認，那么這個任務將分配給其他消費者。

基于這種確認機制，可以在高可靠性和高數據要求情景中，避免數據抓取的遺漏和丟失。

其設計思路應該是基于mq設計兩個接口，一個用于URL的存放，一個用戶URL的獲取，同時基于Redis的URL去重，通過類似scrapy-redis 的調度使爬蟲運行。

主程序示例：

import pika
 
 
class RabbitMQBASE:
 
    def __new__(cls, *args, **kw):
        if not hasattr(cls, '_instance'):
            org = super(RabbitMQBASE, cls)
            cls._instance = org.__new__(cls)
        return cls._instance
 
    def __init__(self, use='root', pwd='111'):
        user_pwd = pika.PlainCredentials(use, pwd)
        self.s_conn = pika.BlockingConnection(
            pika.ConnectionParameters(host='1.1.1.1', heartbeat_interval=3600, credentials=user_pwd))
 
    def channel(self):
        return self.s_conn.channel()
 
    def close(self):
        """關閉連接"""
        self.s_conn.close()
 
    @staticmethod
    def callback(ch, method, properties, body):
        print(" [消費者] %r" % body)
 
 
class RabbitMQ(RabbitMQBASE):
    """
    type_:交換機類型fanout、direct、topic
    exchange：交換機名字
    queue_name：隊列名字,為空則隨機命名
    exclusive：隊列是否持久化False持久,True不持久
    key_list：消費者的交換機、隊列綁定的關鍵詞列表
    key:生產者路由的關鍵詞
    no_ack：是否確認消息True不確定,False確定
    """
    def __init__(self, use='root', pwd='Kw7pGR4xDD1CsP*U', type_='direct', exchange='test',
                 queue_name=None, exclusive=True, key_list=['test'], key='test', no_ack=True):
        RabbitMQBASE.__init__(self, use=use, pwd=pwd)
        self.type_ = type_
        self.exchange = exchange
        self.queue_name = queue_name
        self.exclusive = exclusive
        self.key = key
        self.key_list = key_list
        self.no_ack = no_ack
 
    def rabbit_get(self):
        """消費者"""
        channel = self.channel()
        channel.exchange_declare(exchange=self.exchange, exchange_type=self.type_)
        if self.queue_name == None:
            result = channel.queue_declare(exclusive=self.exclusive)
            self.queue_name = result.method.queue
        if self.type_ != 'fanout':
            for key in self.key_list:
                channel.queue_bind(exchange=self.exchange,  # 將交換機、隊列、關鍵字綁定
                                   queue=self.queue_name, routing_key=key)
        channel.basic_consume(RabbitMQBASE.callback, queue=self.queue_name, no_ack=self.no_ack)
        channel.start_consuming()
 
    def rabbit_put(self, message='hello word'):
        """生產者"""
        channel = self.channel()
        channel.exchange_declare(exchange=self.exchange, exchange_type=self.type_)
        if self.type_ == 'fanout':
            self.key = ""
        channel.basic_publish(exchange=self.exchange, routing_key=self.key, body=message)
        channel.close()

以上是“RabbitMQ怎么在python分布式爬蟲中構建”這篇文章的所有內容，感謝各位的閱讀！希望分享的內容對大家有幫助，更多相關知識，歡迎關注億速云行業資訊頻道！