快捷導(dǎo)航

Python爬蟲(chóng)程序中使用生產(chǎn)者與消費(fèi)者模式時(shí)進(jìn)程過(guò)早退出的問(wèn)題

更新時(shí)間：2023年01月06日 10:34:18 作者：smart_cat

本文主要介紹了Python爬蟲(chóng)程序中使用生產(chǎn)者與消費(fèi)者模式時(shí)進(jìn)程過(guò)早退出的問(wèn)題，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

之前寫(xiě)爬蟲(chóng)程序的時(shí)候，采用生產(chǎn)者和消費(fèi)者的模式，利用Queue作為生產(chǎn)者進(jìn)程和消費(fèi)者進(jìn)程之間的同步隊(duì)列。

執(zhí)行程序時(shí)，總是秒退，加了斷點(diǎn)也無(wú)法中斷，加打印也無(wú)法輸出，我知道肯定是進(jìn)程退出了，但還是百思不得解，為什么會(huì)這么快就退出。

一開(kāi)始以為是我的進(jìn)程代碼寫(xiě)的有問(wèn)題，在某個(gè)地方崩潰導(dǎo)致程序提前退出，排查了一遍又一遍，并沒(méi)有發(fā)現(xiàn)什么明顯的問(wèn)題，后來(lái)走讀代碼，看到主模塊中消費(fèi)者和生產(chǎn)者進(jìn)程的啟動(dòng)后，發(fā)現(xiàn)了問(wèn)題，原因是我通過(guò)start()方法啟動(dòng)進(jìn)程后，使用join()的方式有問(wèn)題。消費(fèi)者進(jìn)程必須執(zhí)行join()操作，否則消費(fèi)者進(jìn)程將在有時(shí)間完成所有工作之前被終止。

錯(cuò)誤的示范：

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

producer = PageContentProducer(queue)
producer.start()

# 想通過(guò)queue的join()方法確保queue中的元素都被處理完畢
# 但從實(shí)際運(yùn)行看，消費(fèi)者進(jìn)程還沒(méi)來(lái)得及處理就退出了
queue.join()

正確的示范：

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

producer = PageContentProducer(queue)
producer.start()
# 需要執(zhí)行producer.join()，確保生產(chǎn)者進(jìn)程能夠持續(xù)執(zhí)行
producer.join()

# 需要執(zhí)行consumer.join()，確保消費(fèi)者進(jìn)程有時(shí)間進(jìn)行處理
consumer.join()

# 通過(guò)queue的join()方法確保queue中的元素都被處理完畢, 這一步可選，因?yàn)檎鎸?shí)代碼里放了隊(duì)列完成標(biāo)志
queue.join()

生產(chǎn)者進(jìn)程示意代碼：

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 = [] ?# 用于保存匯總信息，沒(méi)有什么實(shí)際作用
? ? ? ? self.output_queue = output_queue
? ??
? ? def run(self):
? ? ? ? '''
? ? ? ? 向隊(duì)列中加入每一篇文章
? ? ? ? '''
? ? ? ? self.visit_all_page_to_get_content()
? ? ? ??
? ? def visit_all_page_to_get_content(self):
? ? ? ? '''
? ? ? ? 使用線程池處理所有的page, 并從每一頁(yè)上提取所有的文章content
? ? ? ? '''
? ? ? ? ...

消費(fèi)者進(jìn)程示意代碼：

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:
? ? ? ? ? ? ? ? ? ? # 如果收到結(jié)束標(biāo)志, 就退出當(dāng)前任務(wù)
? ? ? ? ? ? ? ? ? ? break
? ? ? ? ? ? ? ? self.content_worker_func(self.dir, content)
? ? ? ? ? ? ? ? print(f"已處理: {content['title']}")
? ? ? ? ? ? ? ? # 發(fā)出信號(hào)通知任務(wù)完成
? ? ? ? ? ? ? ? self.input_queue.task_done()
? ? ? ? ? ? except Exception as e:
? ? ? ? ? ? ? ? print(repr(e))
? ? ? ? ? ? ? ??
? ? def content_worker_func(self, dir, content):
? ? ? ? '''
? ? ? ? 主要工作函數(shù)
? ? ? ? '''
? ? ? ? ...

主模塊代碼示意如下:

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()
? ??
? ? # 在隊(duì)列上放置標(biāo)志，發(fā)出完成信號(hào), 有幾個(gè)消費(fèi)者,就需要放置多少個(gè)標(biāo)志
? ? for i in range(0, consumer_num):
? ? ? ? queue.put(None)
? ? ? ??
? ? # 等待消費(fèi)者進(jìn)程關(guān)閉
? ? for i in range(0, consumer_num):
? ? ? ? consumers[i].join()

到此這篇關(guān)于Python爬蟲(chóng)程序中使用生產(chǎn)者與消費(fèi)者模式時(shí)進(jìn)程過(guò)早退出的問(wèn)題的文章就介紹到這了,更多相關(guān)Python生產(chǎn)者與消費(fèi)者模式進(jìn)程早退內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: