亲宝软件园·资讯

展开

Python爬虫程序中使用生产者与消费者模式时进程过早退出的问题

smart_cat 人气:0

之前写爬虫程序的时候,采用生产者和消费者的模式,利用Queue作为生产者进程和消费者进程之间的同步队列。

执行程序时,总是秒退,加了断点也无法中断,加打印也无法输出,我知道肯定是进程退出了,但还是百思不得解,为什么会这么快就退出。

一开始以为是我的进程代码写的有问题,在某个地方崩溃导致程序提前退出,排查了一遍又一遍,并没有发现什么明显的问题,后来走读代码,看到主模块中消费者和生产者进程的启动后,发现了问题,原因是我通过start()方法启动进程后,使用join()的方式有问题。消费者进程必须执行join()操作,否则消费者进程将在有时间完成所有工作之前被终止。

错误的示范:

queue = multiprocessing.JoinableQueue()
consumer = PageContentConsumer(queue)
consumer.start()

producer = PageContentProducer(queue)
producer.start()

# 想通过queue的join()方法确保queue中的元素都被处理完毕
# 但从实际运行看,消费者进程还没来得及处理就退出了
queue.join()

正确的示范:

queue = multiprocessing.JoinableQueue()
consumer = PageContentConsumer(queue)
consumer.start()

producer = PageContentProducer(queue)
producer.start()
# 需要执行producer.join(),确保生产者进程能够持续执行
producer.join()

# 需要执行consumer.join(),确保消费者进程有时间进行处理
consumer.join()

# 通过queue的join()方法确保queue中的元素都被处理完毕, 这一步可选,因为真实代码里放了队列完成标志
queue.join()

生产者进程示意代码:

class PageContentProducer(multiprocessing.Process):
    def __init__(self, page_list:list, output_queue:multiprocessing.JoinableQueue):
        multiprocessing.Process.__init__(self)
        self.daemon = True
        self.page_list = page_list
        self.content_list = []  # 用于保存汇总信息,没有什么实际作用
        self.output_queue = output_queue
    
    def run(self):
        '''
        向队列中加入每一篇文章
        '''
        self.visit_all_page_to_get_content()
        
    def visit_all_page_to_get_content(self):
        '''
        使用线程池处理所有的page, 并从每一页上提取所有的文章content
        '''
        ...

消费者进程示意代码:

class PageContentConsumer(multiprocessing.Process):
    def __init__(self, dir, input_queue:multiprocessing.JoinableQueue):
        multiprocessing.Process.__init__(self)
        self.daemon = True
        self.input_queue = input_queue
        self.dir = dir
        
    def run(self):
        while True:
            try:
                content = self.input_queue.get()
                if content is None:
                    # 如果收到结束标志, 就退出当前任务
                    break
                self.content_worker_func(self.dir, content)
                print(f"已处理: {content['title']}")
                # 发出信号通知任务完成
                self.input_queue.task_done()
            except Exception as e:
                print(repr(e))
                
    def content_worker_func(self, dir, content):
        '''
        主要工作函数
        '''
        ...

主模块代码示意如下:

if __name__ == '__main__':
    page_list = [xxxx]
    
    queue = multiprocessing.JoinableQueue()
    
    consumer_num = os.cpu_count()
    consumers = []
    
    for i in range(0, consumer_num):
        consumers.append(PageContentConsumer(dir, queue))
    
    for i in range(0, consumer_num):
        consumers[i].start()
    
    producer = PageContentProducer(page_list, queue)
    producer.start()
    producer.join()
    
    # 在队列上放置标志,发出完成信号, 有几个消费者,就需要放置多少个标志
    for i in range(0, consumer_num):
        queue.put(None)
        
    # 等待消费者进程关闭
    for i in range(0, consumer_num):
        consumers[i].join()

加载全部内容

相关教程
猜你喜欢
用户评论