Python爬蟲之queue線程安全實戰

1.普通下載

import requests
import os
import re
from lxml import etree
from urllib import request


def get_detail(url):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3554.0 Safari/537.36"
    }
    rep = requests.get(url, headers=headers)
    html = etree.HTML(rep.text)
    imgs = html.xpath('//div[@class="page-content text-center"]//img[@class!="gif"]')
    for img in imgs:
        img_url = img.get("data-original")
        # 獲取圖片名稱
        img_name = img.get("alt")
        # 過濾特殊字符
        img_name = re.sub(r'[\?？\.，。！!]', "", img_name)
        # 獲取圖片后綴名
        suffix = os.path.splitext(img_url)[1].split("!")[0]
        filename = img_name + suffix
        # 開始下載到本地
        request.urlretrieve(img_url, "imgs/" + filename)


def main():
    for i in range(1, 101):
        url = "http://www.doutula.com/photo/list/?page={}".format(i)
        get_detail(url)


if __name__ == '__main__':
    main()

2.開啟queue多線程安全隊列異步下載斗圖圖片

import requests
import os
import re
from lxml import etree
from urllib import request
from queue import Queue
import threading


class Producer(threading.Thread):
    """批量下載"""
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3554.0 Safari/537.36"
    }

    def __init__(self, page_queue, img_queue, *args, **kwargs):
        # 找到Producer的父類Thread，然后把Producer的對象self轉換為Thread的對象，調用父類(Thread)的__init__方法，實例化對象
        super(Producer, self).__init__(*args, **kwargs)
        self.page_queue = page_queue
        self.img_queue = img_queue

    def run(self):
        while True:
            # 隊列空為True，則break掉
            if self.page_queue.empty():
                break
            url = self.page_queue.get()
            self.get_detail(url)

    def get_detail(self, url):
        rep = requests.get(url, headers=self.headers)
        text = rep.text
        html = etree.HTML(text)
        imgs = html.xpath('//div[@class="page-content text-center"]//img[@class!="gif"]')
        for img in imgs:
            img_url = img.get("data-original")
            # 獲取圖片名稱
            img_name = img.get("alt")
            # 過濾特殊字符
            img_name = re.sub(r'[\?？\.，。！!\*]', "", img_name)
            # 獲取圖片后綴名
            suffix = os.path.splitext(img_url)[1].split("!")[0]
            filename = img_name + suffix
            # 以元組形式推送到隊列中
            self.img_queue.put((img_url, filename))


class Consumer(threading.Thread):
    """批量存儲"""
    def __init__(self, page_queue, img_queue, *args, **kwargs):
        # 繼承同一個父類，擁有一樣的方法和變量
        super(Consumer, self).__init__(*args, **kwargs)
        self.page_queue = page_queue
        self.img_queue = img_queue

    def run(self):
        while True:
            # 因為是異步下載，所以需要兩個都判斷是否為空
            if self.img_queue.empty() and self.page_queue.empty():
                break
            # 獲取隊列中元組內數據
            img_url, filename = self.img_queue.get()
            request.urlretrieve(img_url, "imgs/" + filename)
            print(filename+"下載完成！")


def main():
    page_queue = Queue(100)       # 設置最大線程數量
    img_queue = Queue(1000)
    for i in range(1, 101):
        url = "http://www.doutula.com/photo/list/?page={}".format(i)
        page_queue.put(url)
    for i in range(5):
        # 開啟五個下載線程
        t = Producer(page_queue, img_queue)
        t.start()

    for x in range(5):
        # 開啟五個儲存線程
        t = Consumer(page_queue, img_queue)
        t.start()


if __name__ == '__main__':
    main()

3.開啟queue多線程安全隊列異步下載百思不得姐段子到本地csv

"""queue安全多線程下載百思不得姐內容到csv"""
import requests
import threading
import csv
from queue import Queue
from lxml import etree


class Producer(threading.Thread):
    """生產者"""
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3554.0 Safari/537.36"
    }

    def __init__(self, page_queue, content_url, *args, **kwargs):
        super(Producer, self).__init__(*args, **kwargs)
        self.page_queue = page_queue
        self.content_url = content_url

    def run(self):
        while True:
            if self.page_queue.empty():
                break
            # 從隊列中獲取url
            url = self.page_queue.get()
            self.get_content(url)

    def get_content(self, url):
        rep = requests.get(url, headers=self.headers)
        text = rep.text
        html = etree.HTML(text)
        contents = html.xpath("//div[@class='j-r-list-c-desc']/a/text()")
        imgs = html.xpath("//div[@class='j-r-list-c-img']//img/@data-original")
        content_tags = zip(contents, imgs)
        for content, img_url in content_tags:
            self.content_url.put((content.strip(), img_url.strip()))


class Consumer(threading.Thread):
    """消費者"""
    def __init__(self, page_queue, content_url, *args, **kwargs):
        super(Consumer, self).__init__(*args, **kwargs)
        self.page_queue = page_queue
        self.content_url = content_url

    def run(self):
        while True:
            if self.page_queue.empty() and self.content_url.empty():
                break
            content, img_url = self.content_url.get()
            # 獲取的內容必須修改成csv支持的格式
            values = {(content, img_url)}
            with open("joke.csv", "a+", encoding="utf-8", newline="") as fp:
                writer = csv.writer(fp)
                # 寫入多個數據
                writer.writerows(values)


def main():
    page_queue = Queue(100)          # 最大線程數量
    content_url = Queue(500)
    for i in range(1, 51):
        url = "http://www.budejie.com/{}".format(i)
        # 推送到隊列中
        page_queue.put(url)
    # 開啟五個線程下載
    for x in range(5):
        t = Producer(page_queue, content_url)
        t.start()
    # 開啟五個線程儲存
    for j in range(5):
        t = Consumer(page_queue, content_url)
        t.start()


if __name__ == '__main__':
    main()