RabbitMQ怎么在python分布式爬蟲中構(gòu)建

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

RabbitMQ是比較靠譜的消息中間件，得益于它的確認(rèn)機(jī)制，當(dāng)一條消息消費(fèi)后如果設(shè)置確定模式，那么確認(rèn)后才會(huì)繼續(xù)消費(fèi)，如果不確定認(rèn)，那么這個(gè)任務(wù)將分配給其他消費(fèi)者。

基于這種確認(rèn)機(jī)制，可以在高可靠性和高數(shù)據(jù)要求情景中，避免數(shù)據(jù)抓取的遺漏和丟失。

其設(shè)計(jì)思路應(yīng)該是基于mq設(shè)計(jì)兩個(gè)接口，一個(gè)用于URL的存放，一個(gè)用戶URL的獲取，同時(shí)基于Redis的URL去重，通過類似scrapy-redis 的調(diào)度使爬蟲運(yùn)行。

主程序示例：

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):
        """關(guān)閉連接"""
        self.s_conn.close()
 
    @staticmethod
    def callback(ch, method, properties, body):
        print(" [消費(fèi)者] %r" % body)
 
 
class RabbitMQ(RabbitMQBASE):
    """
    type_:交換機(jī)類型fanout、direct、topic
    exchange：交換機(jī)名字
    queue_name：隊(duì)列名字,為空則隨機(jī)命名
    exclusive：隊(duì)列是否持久化False持久,True不持久
    key_list：消費(fèi)者的交換機(jī)、隊(duì)列綁定的關(guān)鍵詞列表
    key:生產(chǎn)者路由的關(guān)鍵詞
    no_ack：是否確認(rèn)消息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):
        """消費(fèi)者"""
        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,  # 將交換機(jī)、隊(duì)列、關(guān)鍵字綁定
                                   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'):
        """生產(chǎn)者"""
        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分布式爬蟲中構(gòu)建”這篇文章的所有內(nèi)容，感謝各位的閱讀！希望分享的內(nèi)容對(duì)大家有幫助，更多相關(guān)知識(shí)，歡迎關(guān)注億速云行業(yè)資訊頻道！