這篇文章主要介紹RabbitMQ怎么在python分布式爬蟲中構建,文中介紹的非常詳細,具有一定的參考價值,感興趣的小伙伴們一定要看完!
站在用戶的角度思考問題,與客戶深入溝通,找到蕉嶺網(wǎng)站設計與蕉嶺網(wǎng)站推廣的解決方案,憑借多年的經(jīng)驗,讓設計與互聯(lián)網(wǎng)技術結合,創(chuàng)造個性化、用戶體驗好的作品,建站類型包括:網(wǎng)站建設、網(wǎng)站設計、企業(yè)官網(wǎng)、英文網(wǎng)站、手機端網(wǎng)站、網(wǎng)站推廣、主機域名、虛擬主機、企業(yè)郵箱。業(yè)務覆蓋蕉嶺地區(qū)。
RabbitMQ是比較靠譜的消息中間件,得益于它的確認機制,當一條消息消費后如果設置確定模式,那么確認后才會繼續(xù)消費,如果不確定認,那么這個任務將分配給其他消費者。
基于這種確認機制,可以在高可靠性和高數(shù)據(jù)要求情景中,避免數(shù)據(jù)抓取的遺漏和丟失。
其設計思路應該是基于mq設計兩個接口,一個用于URL的存放,一個用戶URL的獲取,同時基于redis的URL去重,通過類似scrapy-redis 的調(diào)度使爬蟲運行。
主程序示例:
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:生產(chǎn)者路由的關鍵詞 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'): """生產(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分布式爬蟲中構建”這篇文章的所有內(nèi)容,感謝各位的閱讀!希望分享的內(nèi)容對大家有幫助,更多相關知識,歡迎關注創(chuàng)新互聯(lián)行業(yè)資訊頻道!