并發編程專欄系列博客

并發編程（一）python并發編程簡介

并發編程（二）怎樣選擇多線程多進程和多協程

并發編程（三）Python編程慢的罪魁禍首。全局解釋器鎖GIL

并發編程（四）如何使用多線程，使用多線程對爬蟲程序進行修改及比較

并發編程（五）python實現生產者消費者模式多線程爬蟲

并發編程（六）線程安全問題以及lock解決方案

并發編程（七）好用的線程池ThreadPoolExecutor

并發編程（八）在web服務中使用線程池加速

并發編程（九）使用多進程multiprocessing加速程序運行

并發編程（十）在Flask服務中使用進程池加速

并發編程（十一）python異步IO實現并發編程

并發編程（十二）使用subprocess啟動電腦任意程序（聽歌、解壓縮、自動下載等等）

?
?

文章目錄

• • • • • • 多組件的Pipeline技術架構
          • 生產者消費者爬蟲架構
          • 多線程數據通信的queue.Queue
          • 代碼實現

多組件的Pipeline技術架構

• 在介紹或者使用生產者消費者模式前我們先大概了解一下Pipeline架構。

• 例如，對于一件復雜的事情，我們常常都不會一下子做完，而是會分成很多中間步驟一步步完成，進而簡化復雜的事情。請看下圖：

• 如圖，我們現在有一個事情由輸入數據得到輸出數據，中間會經過很多的模塊，而這些模塊之間會通過中間數據進行交互。我們稱這些處理模塊為處理器也叫Processor，而把事情分為很多模塊進行處理的架構稱為Pipeline架構。

• 我們接下來所要用到的生產者消費者模式就是一個典型的Pipeline，他有兩個角色，生產者和消費者。生產者利用輸入數據當做原料生產的結果會通過中間數據傳給消費者進行消費，而消費者則利用中間數據進行消費得到最終的輸出數據。

?

生產者消費者爬蟲架構

• 從上文我們對生產者消費者模式有了簡單了解，接下來我們看一下爬蟲的生產者消費者架構。如圖：

• 從上圖，我們可以看到生產者負責網頁下載，而消費者負責網頁的解析和存儲。兩者之間是互不影響且都可以使用多線程的，那如何把下載好的網頁發給消費者進行處理呢？

  ?

多線程數據通信的queue.Queue

• 我們上文提到多線程的生產者和消費者之間需要數據交互進行完成爬蟲。同時也遺留了一個問題，那就是多線程之間如何進行數據交互。于是，便有了queue.Queue。

• queue.Queue可以用于多線程之間的、線程安全的數據通信。

• 基本語法規則：

  # 1.導入類庫 import queue# 2.創建Queue q = queue.Queue()# 3.添加元素(如果隊列已滿，則進行阻塞，等待隊列不滿時添加) q.put(item)# 4.獲取元素（如果隊列為空，則進行阻塞，等待隊列不空時獲取） item = q.get()# 5.查詢狀態 # 查看元素的多少 q.qsize() # 判斷是否為空 q.empty() # 判斷是否已滿 q.full()

?

代碼實現

• 首先先創建一個下載和解析分開的爬蟲代碼

  # 導包 import requests from bs4 import BeautifulSoup# 爬取鏈接列表 urls = [f"https://w.cnblogs.com/#p{page}"for page in range(1, 51) ]# 獲取下載網頁函數 def craw(url):r = requests.get(url)return r.text# 獲取網頁中的每個博客的鏈接和標題 def parse(html):soup = BeautifulSoup(html, "html.parser")links = soup.find_all("a", class_="post-item-title")return [(link["href"], link.get_text()) for link in links]# 調用函數測試 if __name__ == '__main__':for rst in parse(craw(urls[2])):print(rst)''' output: ('https://www.cnblogs.com/TomDwan/p/14550984.html', '文件上傳漏洞全面滲透姿勢總結') ('https://www.cnblogs.com/ice-image/p/14554250.html', 'RPC 框架設計') ('https://www.cnblogs.com/huaweiyun/p/14554253.html', '都在講Redis主從復制原理，我來講實踐總結') ('https://www.cnblogs.com/JulianHuang/p/14554245.html', 'Ingress-nginx工作原理和實踐') ('https://www.cnblogs.com/tencent-cloud-native/p/14554124.html', '一文讀懂SuperEdge拓撲算法') ('https://www.cnblogs.com/root429/p/14554028.html', '阿里一面CyclicBarrier和CountDownLatch的區別是啥') ('https://www.cnblogs.com/light0011/p/14554003.html', '圖的最短路徑之楊貴妃荔枝自由') ('https://www.cnblogs.com/chiaki/p/14551882.html', '淺析MyBatis（三）：聊一聊MyBatis的實用插件與自定義插件') ('https://www.cnblogs.com/technology-blog/p/14544096.html', 'Spring源碼之ApplicationContext') ('https://www.cnblogs.com/zgrey/p/14553606.html', 'KVM虛擬化配置') ('https://www.cnblogs.com/xiyuanMore/p/14553598.html', '聊聊自驅團隊的構建（四）') ('https://www.cnblogs.com/pengdai/p/14553561.html', 'Tomcat詳解系列(2) - 理解Tomcat架構設計') ('https://www.cnblogs.com/onlyblues/p/14550828.html', '遍歷二叉樹的遞歸與非遞歸代碼實現') ('https://www.cnblogs.com/xueweihan/p/14553532.html', '詳解 ZooKeeper 數據持久化') ('https://www.cnblogs.com/charlieroro/p/14439895.html', '混合云中的事件驅動架構') ('https://www.cnblogs.com/vchar/p/14347260.html', 'Docker上安裝Redis') ('https://www.cnblogs.com/danvic712/p/simplify-abp-generated-template-structure.html', 'ABP 適用性改造 - 精簡 ABP CLI 生成的項目結構') ('https://www.cnblogs.com/flydean/p/14553206.html', '密碼學系列之:csrf跨站點請求偽造') ('https://www.cnblogs.com/cmt/p/14553189.html', '網站整改公告') ('https://www.cnblogs.com/pingguo-softwaretesting/p/14535112.html', '【小白學算法】5.鏈表(linked list)，鏈表的插入、讀取') '''

• 接下來使用多線程實現生產者消費者模式。

  # -*- coding: utf-8 -*- # @Time : 2021-03-20 16:02:55 # @Author : wlq # @FileName: blog_spider1.py # @Email ：rd_wlq@163.com # 導包 import requests from bs4 import BeautifulSoup import queue import time import random import threading# 爬取鏈接列表 urls = [f"https://w.cnblogs.com/#p{page}"for page in range(1, 51) ]# 獲取下載網頁函數 def craw(url):r = requests.get(url)return r.text# 解析網頁函數 def parse(html):soup = BeautifulSoup(html, "html.parser")links = soup.find_all("a", class_="post-item-title")return [(link["href"], link.get_text()) for link in links]# 生產者 def do_craw(url_queue: queue.Queue, html_queue: queue.Queue):while True:url = url_queue.get()html = craw(url)html_queue.put(html)print(threading.current_thread().name, f"craw{url}", "url_queue.size=", url_queue.qsize())time.sleep(random.randint(1, 2))# 消費者 def do_parse(html_queue: queue.Queue, file):while True:html = html_queue.get()rst = parse(html)for x in rst:file.write(str(x) + "\n")print(threading.current_thread().name, "rst.size", len(rst), "html_queue.size=", html_queue.qsize())time.sleep(random.randint(1, 2))if __name__ == '__main__':# 創建輸入數據隊列url_queue = queue.Queue()# 創建中間數據隊列html_queue = queue.Queue()# 將需要爬取的鏈接添加到輸入數據隊列for url in urls:url_queue.put(url)# 創建三個生產者進行生產for idx in range(3):t = threading.Thread(target=do_craw, args=(url_queue, html_queue), name=f"craw{idx}")t.start()file = open('spider.txt', 'w')# 創建兩個消費者消費for idx in range(2):t = threading.Thread(target=do_parse, args=(html_queue, file), name=f"parse{idx}")t.start()''' output: craw2 crawhttps://w.cnblogs.com/#p3 url_queue.size= 47 craw0 crawhttps://w.cnblogs.com/#p1 url_queue.size= 47 parse0 rst.size 20 html_queue.size= 0 parse1 rst.size 20 html_queue.size= 0 craw1 crawhttps://w.cnblogs.com/#p2 url_queue.size= 47 parse0 rst.size 20 html_queue.size= 0 craw2 crawhttps://w.cnblogs.com/#p4 url_queue.size= 46 parse1 rst.size 20 html_queue.size= 0 craw0 crawhttps://w.cnblogs.com/#p5 url_queue.size= 44 parse0 rst.size 20 html_queue.size= 0 craw1 crawhttps://w.cnblogs.com/#p6 url_queue.size= 44 parse1 rst.size 20 html_queue.size= 0 craw0 crawhttps://w.cnblogs.com/#p7 url_queue.size= 42 craw2 crawhttps://w.cnblogs.com/#p8 url_queue.size= 42 parse0 rst.size 20 html_queue.size= 1 parse1 rst.size 20 html_queue.size= 0 craw1 crawhttps://w.cnblogs.com/#p9 url_queue.size= 41 parse1 rst.size 20 html_queue.size= 0 craw0 crawhttps://w.cnblogs.com/#p10 url_queue.size= 38 craw2 crawhttps://w.cnblogs.com/#p11 url_queue.size= 38 craw1 crawhttps://w.cnblogs.com/#p12 url_queue.size= 38 parse0 rst.size 20 html_queue.size= 2 parse1 rst.size 20 html_queue.size= 1 craw0 crawhttps://w.cnblogs.com/#p13 url_queue.size= 35 craw2 crawhttps://w.cnblogs.com/#p14 url_queue.size= 35 craw1 crawhttps://w.cnblogs.com/#p15 url_queue.size= 35 parse0 rst.size 20 html_queue.size= 3 parse1 rst.size 20 html_queue.size= 2 craw0 crawhttps://w.cnblogs.com/#p16 url_queue.size= 32 craw2 crawhttps://w.cnblogs.com/#p17 url_queue.size= 32 craw1 crawhttps://w.cnblogs.com/#p18 url_queue.size= 32 craw0 crawhttps://w.cnblogs.com/#p19 url_queue.size= 30 craw2 crawhttps://w.cnblogs.com/#p20 url_queue.size= 30 parse0 rst.size 20 html_queue.size= 6 parse1 rst.size 20 html_queue.size= 5 craw1 crawhttps://w.cnblogs.com/#p21 url_queue.size= 28 craw0 crawhttps://w.cnblogs.com/#p22 url_queue.size= 28 parse0 rst.size 20 html_queue.size= 6 parse1 rst.size 20 html_queue.size= 5 craw2 crawhttps://w.cnblogs.com/#p23 url_queue.size= 27 parse1 rst.size 20 html_queue.size= 5 craw1 crawhttps://w.cnblogs.com/#p24 url_queue.size= 25 craw0 crawhttps://w.cnblogs.com/#p25 url_queue.size= 25 parse0 rst.size 20 html_queue.size= 6 craw2 crawhttps://w.cnblogs.com/#p26 url_queue.size= 23 parse1 rst.size 20 html_queue.size= 6 craw0 crawhttps://w.cnblogs.com/#p27 url_queue.size= 23 parse0 rst.size 20 html_queue.size= 6 craw1 crawhttps://w.cnblogs.com/#p28 url_queue.size= 21 craw2 crawhttps://w.cnblogs.com/#p29 url_queue.size= 21 craw0 crawhttps://w.cnblogs.com/#p30 url_queue.size= 20 parse1 rst.size 20 html_queue.size= 8 craw0 crawhttps://w.cnblogs.com/#p31 url_queue.size= 19 parse0 rst.size 20 html_queue.size= 8 craw1 crawhttps://w.cnblogs.com/#p32 url_queue.size= 17 craw2 crawhttps://w.cnblogs.com/#p33 url_queue.size= 17 craw0 crawhttps://w.cnblogs.com/#p34 url_queue.size= 16 parse0 rst.size 20 html_queue.size= 10 parse1 rst.size 20 html_queue.size= 9 craw1 crawhttps://w.cnblogs.com/#p35 url_queue.size= 15 craw0 crawhttps://w.cnblogs.com/#p36 url_queue.size= 14 parse1 rst.size 20 html_queue.size= 10 craw2 crawhttps://w.cnblogs.com/#p37 url_queue.size= 13 parse0 rst.size 20 html_queue.size= 10 craw1 crawhttps://w.cnblogs.com/#p38 url_queue.size= 12 craw0 crawhttps://w.cnblogs.com/#p39 url_queue.size= 11 parse1 rst.size 20 html_queue.size= 11 craw2 crawhttps://w.cnblogs.com/#p40 url_queue.size= 9 craw1 crawhttps://w.cnblogs.com/#p41 url_queue.size= 9 parse0 rst.size 20 html_queue.size= 12 craw0 crawhttps://w.cnblogs.com/#p42 url_queue.size= 8 parse0 rst.size 20 html_queue.size= 12 parse1 rst.size 20 html_queue.size= 11 craw2 crawhttps://w.cnblogs.com/#p43 url_queue.size= 6 craw1 crawhttps://w.cnblogs.com/#p44 url_queue.size= 6 parse1 rst.size 20 html_queue.size= 12 craw0 crawhttps://w.cnblogs.com/#p45 url_queue.size= 5 craw2 crawhttps://w.cnblogs.com/#p46 url_queue.size= 4 parse0 rst.size 20 html_queue.size= 13 parse1 rst.size 20 html_queue.size= 12 craw0 crawhttps://w.cnblogs.com/#p47 url_queue.size= 3 craw1 crawhttps://w.cnblogs.com/#p48 url_queue.size= 1 craw2 crawhttps://w.cnblogs.com/#p49 url_queue.size= 1 craw0 crawhttps://w.cnblogs.com/#p50 url_queue.size= 0 parse0 rst.size 20 html_queue.size= 15 parse1 rst.size 20 html_queue.size= 14 parse0 rst.size 20 html_queue.size= 13 parse1 rst.size 20 html_queue.size= 12 parse0 rst.size 20 html_queue.size= 11 parse1 rst.size 20 html_queue.size= 10 parse0 rst.size 20 html_queue.size= 9 parse1 rst.size 20 html_queue.size= 8 parse1 rst.size 20 html_queue.size= 7 parse0 rst.size 20 html_queue.size= 6 parse1 rst.size 20 html_queue.size= 5 parse0 rst.size 20 html_queue.size= 4 parse0 rst.size 20 html_queue.size= 3 parse1 rst.size 20 html_queue.size= 2 parse0 rst.size 20 html_queue.size= 1 parse0 rst.size 20 html_queue.size= 0 '''

  通過輸出可以看出，最開始url_queue為47，是因為一開始三個生產者已經拿走三個。接著用過生產者的不斷生產，輸入數據隊列逐漸變小。而中間數據隊列html_queue在一會兒變大一會兒變小，最后一部分也是逐漸變小。這是因為剛開始中間隊列為空，生產者生產完成后會將數據存入中間隊列，導致中間隊列變大，消費者在監測到中間隊列不為空時，取出中間數據進行消費導致隊列變小，兩者同時進行導致中間隊列大小的波動。最后，生產者隊列所有數據取完，生產者不再生產數據，中間隊列不再變大。同時，消費者繼續消費，導致中間隊列持續變小，直至為空。

  注:最后程序一直沒有結束是因為生產者和消費者一直監測生產隊列和消費隊列是否有新的數據出現。

總結

以上是生活随笔為你收集整理的并发编程（五）python实现生产者消费者模式多线程爬虫的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。