Python爬虫编程7——多线程爬虫

2023-05-16

一.多线程基本介绍

程序中模拟多任务

二.多线程的创建

三.主线程与子线程的执行关系

四.查看线程数量

五.线程间的通信（多线程共享全局变量）

六.线程间的资源竞争

互斥锁和死锁

互斥锁

死锁

七.Queue线程

八.生产者和消费者

Lock版的生产者和消费者

Condition版的生产者和消费者

九.多线程的应用实例

爬取小米商城使用普通方式爬取

使用多线程爬取

一.多线程基本介绍

有很多场景中的事情是同时进行的，比如开车的时候手和脚来共同驾驶汽车，再比如唱歌跳舞也是同时进行的。

程序中模拟多任务

import time

def sing():
    for i in range(3):
        print("正在唱歌...%d"%i)
        time.sleep(1)

def dance():
    for i in range(3):
        print("正在跳舞...%d"%i)
        time.sleep(1)

if __name__ == '__main__':
    sing()
    dance()

二.多线程的创建

当调用Thread的时候，不会创建线程。

当调用Thread创建出来的实例对象的start方法的时候，才会创建线程以及开始运行这个线程。

（1）通过函数来创建多线程：

import threading
import time


# 通过函数创建多线程
def demo():
    # 线程的函数事件
    print('子线程！')


if __name__ == '__main__':
    for i in range(8):
        t = threading.Thread(target=demo)  # 只是创建还没有启动
        t.start()  # 启动(一个可以启动的状态)

（2）通过类来创建多线程：

import threading
import time

# 通过类来创建多线程
class MyThread(threading.Thread):
    # 重写run方法
    def run(self):
        for i in range(5):
            print("这是一个子线程！")


if __name__ == '__main__':
    m = MyThread()
    # start 启动一个子线程
    m.start()

（3）对线程中传入参数时，需要使用参数 args 注意传入的是一个元组；

    t1 = threading.Thread(target=demo1, args=(100000,))
    t2 = threading.Thread(target=demo2, args=(100000,))

三.主线程与子线程的执行关系

（1）主线程会等待子线程结束之后结束；

（2）join（）等待子线程结束之后，主线程继续执行；

（3）setDaemon（）守护线程，不会等待子线程结束，即当主线程执行完，无论子线程有没有技术都会结束程序；

import threading
import time

def text():
    for i in range(4):
        print("子线程！")
        # time.sleep(1)


if __name__ == '__main__':
    t = threading.Thread(target=text)
    t.start()
    # 强制等待
    # time.sleep(3)
    # t.setDaemon()   # 当主线程运行完，无论子线程玩没玩都会结束
    t.join()  # 不论子线程运行多久，一定是要等待子线程运行完成之后才运行主线程，有个timeout参数，可以设置等待时间
    print("111")

四.查看线程数量

（1）enumerate() 方法在循环中使用时，会连同索引一起返回：

# enumerate()  连同索引一起返回
text_list = ['xxx', 'yyy', 'zzz']
for index, i in enumerate(text_list):
    # print(type(i), i)
    print(index, i)

（2）threading.enumerate() 查看线程数量的方法：

threading.enumerate()	查看当前线程的数量

使用例子：

import threading
import time


# threading.enumerate()
def demo1():
    for i in range(8):
        time.sleep(1)
        print(f'demo1--{i}')


def demo2():
    for i in range(5):
        time.sleep(1)
        print(f'demo2--{i}')


if __name__ == '__main__':
    t1 = threading.Thread(target=demo1)
    # print(threading.enumerate())
    t2 = threading.Thread(target=demo2)
    t1.start()
    # 在start开始时子线程才会创建成功
    # print(threading.enumerate())
    t2.start()
    # print(threading.enumerate())
    while True:
        time.sleep(1)
        print(threading.enumerate())
        if len(threading.enumerate()) <= 1:
            break

五.线程间的通信（多线程共享全局变量）

在一个函数中，对全局变量进行修改的时候，是否要加 global 要看是否对全局变量的指向进行了修改，如果修改了指向，那么必须使用 global ，仅仅是修改了指向的空间中的数据，此时不用必须使用 global ，线程是共享全局变量。

六.线程间的资源竞争

一个线程写入，一个线程读取，没问题，如果两个线程都写入呢？

互斥锁和死锁

互斥锁

当多个线程几乎同时修改某一个共享数据的时候，需要进行同步控制。

某个线程要更改共享数据时，先将其锁定，此时资源的状态为锁定，其他线程不能改变，只能该线程释放资源，将资源的状态变成非锁定，其它的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。

创建锁
mutex = threading.Lock()

锁定
mutex.acquire()

解锁
mutex.release()

死锁

在线程共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方资源，就会造成死锁。

import threading
import time

class MyThread1(threading.Thread):
    def run(self):
        # 对mutexA上锁
        mutexA.acquire()

        # mutexA上锁后，延时1秒，等待另外那个线程 把mutexB上锁
        print(self.name+'----do1---up----')
        time.sleep(1)

        # 此时会堵塞，因为这个mutexB已经被另外的线程抢先上锁了
        mutexB.acquire()
        print(self.name+'----do1---down----')
        mutexB.release()

        # 对mutexA解锁
        mutexA.release()

class MyThread2(threading.Thread):
    def run(self):
        # 对mutexB上锁
        mutexB.acquire()

        # mutexB上锁后，延时1秒，等待另外那个线程 把mutexA上锁
        print(self.name+'----do2---up----')
        time.sleep(1)

        # 此时会堵塞，因为这个mutexA已经被另外的线程抢先上锁了
        mutexA.acquire()
        print(self.name+'----do2---down----')
        mutexA.release()

        # 对mutexB解锁
        mutexB.release()

mutexA = threading.Lock()
mutexB = threading.Lock()

if __name__ == '__main__':
    t1 = MyThread1()
    t2 = MyThread2()
    t1.start()
    t2.start()

实例：

import threading
import time

num = 100
# 线程锁，解决资源竞争问题
lock = threading.Lock()
# RLock 可以上多把锁,上多少解多少
# rlock = threading.RLock()

def demo1(num1):
    global num
    # 上锁
    lock.acquire()
    for i in range(num1):
        num += 1
    # 解锁
    lock.release()
    print(f'demo1--{num}')


def demo2(num1):
    global num
    lock.acquire()
    for i in range(num1):
        num += 1
    lock.release()
    print(f'demo2--{num}')


def main():
    # 使用 args 传参  当传入的参数很大时，资源竞争更加明显，CPU调度问题
    t1 = threading.Thread(target=demo1, args=(100000,))
    t2 = threading.Thread(target=demo2, args=(100000,))
    t1.start()
    t2.start()
    print(f'main--{num}')


if __name__ == '__main__':
    main()

七.Queue线程

在线程中，访问一些全局变量，加锁是一个经常的过程。如果你是想把一些数据存储到某个队列中，那么python内置了一个线程安全的模块叫做queue模块。python中的queue模块中提供了同步的，线程安全的队列类，包括FIFO（先进先出）队列Queue，LIFO（后入先出）队列LifoQueue。这些队列都实现了锁原语（可以理解为原子操作，即要么不做，要么都做完），能够在多线程中直接使用。可以使用队列来实现线程间的同步。

初始化Queue(maxsize)：创建一个先进先出的队列。
empty()：判断队列是否为空。
full()：判断队列是否满了。
get()：从队列中取最后一个数据。
put()：将一个数据放到队列中。

使用实例：


from queue import Queue

'''
四种常用的队列方法
empty()：判断队列是否为空
full()：判断队列是否满了
get()：从队列中取最后一个数据
put()：把一个数据放到队列中
'''

q = Queue(3)   # 创建时可以传入队列的大小
# 判断队列是否为空，返回的是布尔值，True表示为空，False表示不为空
print(q.empty())
print(q.full())
q.put(1)
q.put(2)
q.put(3)
# print(q.put_nowait(2))  #立即向队列添加，否则报错

# 先进先出
print(q.get())
print(q.get())
print(q.get())
# print(q.get(timeout=2))
# print(q.get_nowait(2))  #立即向队列取出，否则报错

print('*' * 50)
print(q.empty())
# 判断队列是否为满，返回的是布尔值，True表示满了，False表示不满
print(q.full())

# 当前队列大小
# print(q.qsize())

# q.put(4, timeout=2)
# q.put_nowait(4)

八.生产者和消费者

生产者和消费者模式是多线程开发中常见的一种模式。通过生产者和消费者模式，可以让代码达到高内聚低耦合的目标，线程管理更加方便，程序分工更加明确。

生产者的线程专门用来生产一些数据，然后存放到容器中(中间变量)。消费者在从这个中间的容器中取出数据进行消费。

Lock版的生产者和消费者

import threading
import random
gMoney = 0
# 定义一个变量 保存生产的次数 默认是0次
gTimes = 0
# 定义一把锁
gLock = threading.Lock()

# 定义生产者
class Producer(threading.Thread):

    def run(self):
        global gMoney
        global gTimes
        gLock.acquire()  # 上锁
        while True:
            # gLock.acquire() # 上锁
            if gTimes >= 10:
                # gLock.release()
                break
            money = random.randint(0,100)
            gMoney += money
            gTimes += 1
            print("%s生产了%d元钱" % (threading.current_thread().name, money))
        gLock.release() # 解锁
# 定义消费者
class Consumer(threading.Thread)
    def run(self):
        global gMoney
        while True:
            gLock.acquire()  # 上锁
            money = random.randint(0, 100)
            if gMoney >= money:
                gMoney -= money
                print("%s消费了%d元钱" % (threading.current_thread().name, money))
            else:
                if gTimes >= 10:
                    gLock.release()
                    break
                print("%s想消费%d元钱，但是余额只有%d"%(threading.current_thread().name,money,gMoney))
            gLock.release()  # 解锁

def main():
    # 开启5个生产者线程
    for x in range(5):
        th = Producer(name="生产者%d号" % x)
        th.start()
    # 开启5个消费者线程
    for x in range(5):
        th = Consumer(name="消费者%d号" % x)
        th.start()

if __name__ == '__main__':
    main()

Condition版的生产者和消费者

import threading
import random
gMoney = 0
# 定义一个变量 保存生产的次数 默认是0次
gTimes = 0
# 定义一把锁
# gLock = threading.Lock()
gCond = threading.Condition()
# 定义生产者
class Producer(threading.Thread):

    def run(self):
        global gMoney
        global gTimes

        while True:
            gCond.acquire() # 上锁
            if gTimes >= 10:
                gCond.release()
                break
            money = random.randint(0,100)
            gMoney += money
            gTimes += 1
            print("%s生产了%d元钱，剩余%d元钱" % (threading.current_thread().name, money, gMoney))
            gCond.notify_all()
            gCond.release() # 解锁
# 定义消费者
class Consumer(threading.Thread):

    def run(self):
        global gMoney
        while True:
            gCond.acquire()  # 上锁
            money = random.randint(0, 100)

            while gMoney < money:
                if gTimes >= 10:
                    gCond.release()
                    return  # 这里如果用break只能退出外层循环，所以我们直接return
                print("%s想消费%d元钱，但是余额只有%d元钱了，并且生产者已经不再生产了！"%(threading.current_thread().name,money,gMoney))
                gCond.wait()
            # 开始消费
            gMoney -= money
            print("%s消费了%d元钱，剩余%d元钱" % (threading.current_thread().name, money, gMoney))
            gCond.release()

def main():
    # 开启5个生产者线程
    for x in range(5):
        th = Producer(name="生产者%d号" % x)
        th.start()
    # 开启5个消费者线程
    for x in range(5):
        th = Consumer(name="消费者%d号" % x)
        th.start()

if __name__ == '__main__':
    main()

九.多线程的应用实例

爬取小米商城使用普通方式爬取

import requests
import time
import json
import pprint


class XiaomiSpider():
    def __init__(self):
        self.url = 'https://app.mi.com/categotyAllListApi?page={}&categoryId=6&pageSize=30'
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36'
        }

    # 发起请求 获取响应内容
    def get_source(self, url):
        response = requests.get(url, headers=self.headers)
        # 得到json数据方案一 通过python内置的json模块
        # result = json.loads(response.text)
        # 得到json数据方案二 通过requests提供的得到json数据的方式
        result = response.json()
        # pprint.pprint(result)
        return result

    # 解析响应内容
    def parse_html(self, html):
        for app_dict in html['data']:
            item = {}
            # pprint.pprint(app_dict)
            item['app_name'] = app_dict['displayName']
            item['app_type'] = app_dict['level1CategoryName']
            item['app_id'] = 'https://app.mi.com/details?id=' + app_dict['packageName']
            print(item)

    def main(self):
        # 进行翻页处理
        for page in range(3):
            new_url = self.url.format(page)
            html = self.get_source(new_url)
            self.parse_html(html)
            time.sleep(1)



if __name__ == '__main__':
    # 开始时间
    start_time = time.time()
    x = XiaomiSpider()
    x.main()
    # 结束时间
    end_time = time.time()
    print('用时%f' % (end_time - start_time))

使用多线程爬取

import requests
import time
import threading
from queue import Queue

class XiaomiSpider():
    def __init__(self):
        self.url = 'https://app.mi.com/categotyAllListApi?page={}&categoryId=6&pageSize=30'
        # 创建队列
        self.q = Queue()
        # 创建锁
        self.lock = threading.Lock()

    # 把目标url放入队列中
    def put_url(self):
        # range(3)--(0,1,2)
        for page in range(3):
            url = self.url.format(page)
            self.q.put(url)

    # 发请求 获响应 解析数据
    def parse_html(self):
        while True:
            self.lock.acquire()
            if not self.q.empty():
                url = self.q.get()
                self.lock.release()
                headers = {
                    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.159 Safari/537.36'
                }
                html = requests.get(url, headers=headers).json()
                # 解析数据
                for app_dict in html['data']:
                    item = {}
                    # pprint.pprint(app_dict)
                    item['app_name'] = app_dict['displayName']
                    item['app_type'] = app_dict['level1CategoryName']
                    item['app_id'] = 'https://app.mi.com/details?id=' + app_dict['packageName']
                    # print(item)

            else:
                self.lock.release()
                break

    def run(self):
        self.put_url()
        # 线程列表
        t_lst = []
        for i in range(2):
            t = threading.Thread(target=self.parse_html)
            t_lst.append(t)
            t.start()


if __name__ == '__main__':
    # 开始时间
    start_time = time.time()
    x = XiaomiSpider()
    x.run()
    # 结束时间
    end_time = time.time()
    print('用时%f' % (end_time - start_time))

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)