這篇文章主要介紹RabbitMQ怎么在python分布式爬蟲中構(gòu)建,文中介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們一定要看完!
白云鄂ssl適用于網(wǎng)站、小程序/APP、API接口等需要進(jìn)行數(shù)據(jù)傳輸應(yīng)用場景,ssl證書未來市場廣闊!成為創(chuàng)新互聯(lián)的ssl證書銷售渠道,可以享受市場價(jià)格4-6折優(yōu)惠!如果有意向歡迎電話聯(lián)系或者加微信:18982081108(備注:SSL證書合作)期待與您的合作!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)注創(chuàng)新互聯(lián)行業(yè)資訊頻道!