多线程爬虫之生产者和消费者模式

2024-03-08 11:25:52

1.什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

2.为什么要使用生产者和消费者模式

在进程世界里，生产者就是生产数据的进程，消费者就是消费数据的进程。在多进程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

3.多线程爬虫

为了提升爬虫运行的效率，我们常常会使用多线程：

点击查看代码

import threading
import requests
import pymongo

# 自定义类
class TencentSpider(threading.Thread):
    # 定义初始化方法
    def __init__(self,page):
        # 处理父类init
        super().__init__()
        self.page = page
   
    # 覆写run方法
    def run(self):
        params['pageIndex'] = self.page
        # 发起请求，接收响应
        response = requests.get(url='https://careers.tencent.com/tencentcareer/api/post/Query', headers=headers,
                                params=params)

        data_list = response.json()['Data']['Posts']
        self.get_data(data_list)
    

    # 定义获取数据函数
    def get_data(self,data_list):
        for data in data_list:
            dic = {}
            # 获取岗位名
            RecruitPostName = data['RecruitPostName']
            # 获取地点
            LocationName = data['LocationName']
            # 获取时间
            LastUpdateTime = data['LastUpdateTime']
            dic['RecruitPostName'] = RecruitPostName
            dic['LocationName'] = LocationName
            dic['LastUpdateTime'] = LastUpdateTime
            # print(dic)
            # 执行插入操作
            col.insert(dic)
            print(RecruitPostName, LocationName, LastUpdateTime)

        pass
    def __del__(self):
        # 关闭数据库
        client.close()

if __name__ == '__main__':
    # 建立连接
    client = pymongo.MongoClient(host='127.0.0.1',port=27017)

    # 3.进入数据库
    db = client['Tencent']

    # 4. 进入集合
    col = db['col']

    # 定义请求头
    headers = {
        'user-agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36'
    }

    # 定义请求参数
    params = {
        'pageSize': '10'
    }

    for page in range(1,11):
        # 创建多线程
        t = TencentSpider(page)
        # 启动多线程
        t.start()

# 线程创建数量公式：
# 单核cpu线程数量 = 1+(I/o利用率÷cpu/利用率)
# 多核cpu线程数量 = cpu核数 * (1+(I/o利用率÷cpu/利用率))  =  2N

运行可以发现，多线程比单线程快的多，但是，任然存在问题，会发现每一请求都会创建一个线程，而在数据量大的情况下可能会有上百次请求，但不可能创建上百个线程去运行程序，一方面这并不会使得我们运行的效率越来越快，反而会造成内存资源的浪费。所以我们提出解决的办法：使用队列。

点击查看代码

import threading
import requests
import pymongo
from queue import Queue

# 自定义类
class TencentSpider(threading.Thread):
    # 定义初始化方法
    def __init__(self,q):
        # 处理父类init
        super().__init__()
        self.q = q
 
    # 覆写run方法
    def run(self):
        # 制定死循环 --- 让队列中的URL能够取空
        while True:
            # 判断队列是否为空
            if self.q.empty():
                break
            # 从队列中取出URL
            url = self.q.get()
            # # 发起请求，接收响应
            response = requests.get(url=url, headers=headers,
                                    params=params)
            data_list = response.json()['Data']['Posts']
            self.get_data(data_list)
   

    # 定义获取数据函数
    def get_data(self,data_list):
        for data in data_list:
            dic = {}
            # 获取岗位名
            RecruitPostName = data['RecruitPostName']
            # 获取地点
            LocationName = data['LocationName']
            # 获取时间
            LastUpdateTime = data['LastUpdateTime']
            dic['RecruitPostName'] = RecruitPostName
            dic['LocationName'] = LocationName
            dic['LastUpdateTime'] = LastUpdateTime
            # 执行插入操作
            col.insert(dic)
            print(RecruitPostName, LocationName, LastUpdateTime)

        pass
    def __del__(self):
        # 关闭数据库
        client.close()

if __name__ == '__main__':
    # 建立连接
    client = pymongo.MongoClient(host='127.0.0.1',port=27017)

    # 3.进入数据库
    db = client['Tencent']

    # 4. 进入集合
    col = db['col']

    # 定义请求头
    headers = {
        'user-agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36'
    }

    # 定义请求参数
    params = {
        'pageSize': '10'
    }

    # 定义基础URL
    base_url = 'https://careers.tencent.com/tencentcareer/api/post/Query?pageSize=10&pageIndex={}'

    # 创建队列
    q = Queue()

    for page in range(1,11):
        q.put(base_url.format(page))


    # 创建3个线程
    for i in range(3):
        # 创建线程
        t = TencentSpider(q)
        # 启动线程
        t.start()

虽然使用队列能够解决创建多线程数量的问题但是代码耦合性太强，一个线程干了太多的活使用。因此我们可以使用生产者和消费者模式去解决该问题

点击查看代码

# threading
# 生产者和消费者  生产者 --- 用于请求    消费者 --- 获取数据保存
# 队列   --- 创建两个队列，一个队列存放请求的URL  另一个队列存放数据

import threading
import requests
import pymongo
from queue import Queue


# 创建生产者类
class Product(threading.Thread):
    # 定义初始化
    def __init__(self,page_queue,data_queue):
        # 处理父类init
        super().__init__()
        self.page_queue = page_queue
        self.data_queue = data_queue
        pass
    # 覆写run方法
    def run(self):
        # 为了保证队列中的URL能够全部取出，写一个死循环
        while True:
            # 判断队列是否为空
            if self.page_queue.empty():
                break
            # 从page_queue中取出URL
            url = self.page_queue.get()
            self.get_content(url)

    # 定义获取数据函数
    def get_content(self,url):
        # 发起请求，接收响应
        data_list = requests.get(url=url,headers=headers).json()['Data']['Posts']
        for data in data_list:
            dic = {}
            # 获取岗位名
            RecruitPostName = data['RecruitPostName']
            # 获取地点
            LocationName = data['LocationName']
            # 获取时间
            LastUpdateTime = data['LastUpdateTime']
            dic['RecruitPostName'] = RecruitPostName
            dic['LocationName'] = LocationName
            dic['LastUpdateTime'] = LastUpdateTime
            # 执行插入操作
            # col.insert(dic)
            self.data_queue.put(dic)
            print(RecruitPostName, LocationName, LastUpdateTime)

# 创建消费者类
class Consumer(threading.Thread):
    # 定义初始化方法
    def __init__(self,data_queue):
        # 处理父类init
        super().__init__()
        self.data_queue = data_queue
      
    # 覆写run方法
    def run(self):
        # 为了保证能够将数据全部取出，写一个死循环
        while True:
            # 需要判断退出条件
            if self.data_queue.empty() and flag==1:
                break
            try:
                # 从data_queue中取出数据
                data = self.data_queue.get(timeout=3)
                self.save(data)
            except:
                pass

    # 定义保存数据函数
    def save(self,data):
        col.insert(data)

    # 关闭数据库
    def __del__(self):
        # 关闭数据库
        client.close()


if __name__ == '__main__':
  
    # 标识位，当生产者生产完时，为1
    flag = 0

    # 建立连接
    client = pymongo.MongoClient(host='127.0.0.1', port=27017)

    # 3.进入数据库
    db = client['Tencent']

    # 4. 进入集合
    col = db['col']

    # 定义请求头
    headers = {
        'user-agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36'
    }

    # 定义基础URL
    base_url = 'https://careers.tencent.com/tencentcareer/api/post/Query?pageSize=10&pageIndex={}'

    # 创建存放URL的队列
    page_queue = Queue()
    for page in range(1,101):
        # 存放URL
        page_queue.put(base_url.format(page))
      

    # 创建存放数据的队列
    data_queue = Queue()

    # 创建一个存放生产者的列表
    p_list = []
    # 创建三个生产者线程
    for i in range(3):
        # 创建线程
        p = Product(page_queue,data_queue)
        p_list.append(p)
        # 启动线程
        p.start()

    # 创建消费者线程
    for i in range(5):
        # 创建线程
        c = Consumer(data_queue)
        # 启动线程
        c.start()
       
    # 线程阻塞，等待生产者线程执行完后flag=True
    for p in p_list:
        p.join()      
    flag = 1

码农公寓

1.什么是生产者消费者模式

2.为什么要使用生产者和消费者模式

相关文章