RabbitMQ怎么在python分布式爬虫中构建
创始人
2024-11-20 20:42:52
0次

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()

相关内容

热门资讯

空调连续工作多久需要停机休息 空调连续工作的时间取决于多个因素,包括空调的型号、功率、使用环境和维护情况等。一般来说,空调可以连续...
空调哪个符号是除湿的 空调除湿功能通常用一个水滴形状的符号表示。不同品牌和型号的空调遥控器可能会有所差异,但一般来说,除湿...
格力空调多长时间加一次氟利昂 1、一般情况下,空调对氟利昂的使用量是非常小的,所以在空调使用了四年左右的时候再考虑加也没问题。2、...
空调用多长时间需要加氟一次 一般情况下是五年。当出现以下这几种情况时,空调才需要加氟。1、空调的室内机和室外机各有两个接口,当接...
格调生活·C标思享荟:卡萨帝在... 9月12日的郑州龙湖金融岛,夜风轻拂,灯光璀璨。“C标时刻共传奇——卡萨帝思享荟暨20周年品牌盛典郑...
品牌价值破千亿、高端份额持续领... 9月12日,“C标时刻共传奇——卡萨帝思享荟暨20周年品牌盛典郑州站”在郑州龙湖金融岛举办。2026...
空调主板坏了修800还是换新的 空调主板坏了是修还是换新,需要考虑多个因素,以下是一些建议:1. 维修成本:如果维修费用较低,比如 ...
空调开多久需要休息多长时间 问题:空调开多久需要休息多长时间回答:1、在使用空调的过程中,一般空调开7~8个小时左右需要休息2~...
中央空调多长时间清洗一次 1、一般空调开机前清洗一次即可,空调开机中间时段再清洗一次,空调关机时清洗一次,这样比较合理。2、空...
中央空调氟利昂多久加一次 看空调制热和制冷程度,很多人的观点认为氟利昂必须一年放一次,其实这个观点是错误的,理解上面有点偏差,...