【Python基础】线程

文章目录

• • @[toc]
  • • 线程与进程的区别与联系
    • 同步任务
    • • 示例
    • 并发任务
    • • 示例
      • 线程调度的“随机性”
    • 线程方法
    • • `thread_object.start()`
      • `thread_object.join()`
      • `thread_object.setDaemon()`
      • • 没有设置守护线程的情况
        • 设置守护线程的情况
      • `thread_object.current_thread()`
    • 目前爬虫的三种实现
    • • 单线程爬虫
      • 多线程并发爬虫
      • 多进程并发爬虫
    • `GIL`锁
    • 线程安全
    • • 不可控的调度
      • 锁
      • • 示例
        • 使用上下文管理器管理锁
      • `RLock`和`Lock`
      • • `Lock`
        • `RLock`
      • 内置类型底层锁机制
      • 死锁
      • • 示例
    • 线程池
    • • 线程池的创建
      • `done()`
      • `cancel()`
      • `result()`
      • `as_completed()`
      • `map()`
      • `wait()`
    • 自定义线程类
    • • 示例

个人主页：丷从心

系列专栏：Python基础

学习指南：Python学习指南

---

线程与进程的区别与联系

• 线程是可以被计算机CPU调度的最小单元
• 进程是计算机分配资源（CPU、内存等）的最小单元
• 一个进程中至少有一个线程，同一个进程中的线程共享进程中的资源

---

同步任务

• 我们在此之前编写的代码都是同步代码，代码从上到下按顺序执行，如果前一个任务没有完成，那么不能运行之后的任务

示例

import time


def work_1():
    print('任务1...')

    time.sleep(2)


def work_2():
    print('任务2...')

    time.sleep(2)


start = time.time()

work_1()
work_2()

end = time.time()

print(f'总共用时: {end - start} s')

任务1...
任务2...
总共用时: 4.020084857940674 s

• 可以看到整个程序用时$4$秒，work_2()需要等待work_1()运行结束后才能运行

---

并发任务

• 使用线程来运行上面的代码，能够优化运行时间

示例

import time
import threading


def work_1():
    print('任务1...')

    time.sleep(2)


def work_2():
    print('任务2...')

    time.sleep(2)


# 通过 Thread 类创建线程对象, 并使用 target 绑定线程对象要运行的任务
t1 = threading.Thread(target=work_1)
t2 = threading.Thread(target=work_2)

# 运行线程
start = time.time()

t1.start()
t2.start()

t1.join()
t2.join()

end = time.time()

print(f'总共用时: {end - start} s')

任务1...
任务2...
总共用时: 2.0165793895721436 s

• 可以看到整个程序用时$2$秒，work_1()和work_2并发运行

线程调度的“随机性”

• 下面的示例可以看到CPU调度线程时的“随机性”

import time
import threading


def work_1():
    for i in range(5):
        print('任务1...')

        time.sleep(2)


def work_2():
    for i in range(5):
        print('任务2...')

        time.sleep(2)


t1 = threading.Thread(target=work_1)
t2 = threading.Thread(target=work_2)

t1.start()
t2.start()

任务1...
任务2...
任务2...任务1...

任务1...任务2...

任务2...
任务1...
任务2...
任务1...

• 可以看到任务$1$和任务$2$的调度顺序是我们无法确定的，是由CPU的调度算法决定的

---

线程方法

• 在学习线程方法之前，我们需要知道Python程序是如何被运行的
  • 一个Python文件被解释器运行时会在操作系统中创建一个进程
  • 然后该进程会创建一个线程来运行文件中的代码，这个程序最初创建的线程称为主线程
  • 当主线程执行到t = threading.Thread()时会创建一个新的线程，称为子线程
  • 当前进程中的主线程与子线程由CPU进行调度，并发地运行，具体如何调度线程由操作系统的调度算法决定
  • 子线程在运行时，主线程不会等待子线程，而是继续向下执行，直到执行到文件末尾没有代码时，主线程会等待子线程运行结束后再退出

thread_object.start()

• t = threading.Thread()只是创建了一个线程，并不会执行线程代码

• t.start()使线程t达到就绪状态，等待CPU进行调度，具体何时调度由CPU决定

• 以上面的并发任务的代码为例，先注释掉t1.join()和t2.join()

import time
import threading


def work_1():
    print('任务1...')

    time.sleep(2)


def work_2():
    print('任务2...')

    time.sleep(2)


# 通过 Thread 类创建线程对象, 并使用 target 绑定线程对象要运行的任务
t1 = threading.Thread(target=work_1)
t2 = threading.Thread(target=work_2)

# 运行线程
start = time.time()

t1.start()
t2.start()

# t1.join()
# t2.join()

end = time.time()

print(f'总共用时: {end - start} s')

任务1...
任务2...
总共用时: 0.0009987354278564453 s

• 可以看到主线程没有等待子线程，而是继续向下执行
• 当执行到end = time.time()时，此时end记录的时间是主线程运行到这行代码的时间
• 之后运行print(f'总共用时: {end - start} s')，输出时间0.0009987354278564453 s，此时执行到了文件末尾没有其他代码，主线程会等待子线程运行结束后再退出
• 为了能正确记录线程运行的时间，我们需要让主线程等待子线程

thread_object.join()

• t.join()使主线程等待子线程，子线程任务执行结束后主线程再继续向下执行
• 仍然以上面的并发任务的代码为例，取消注释t1.join()和t2.join()

import time
import threading


def work_1():
    print('任务1...')

    time.sleep(2)


def work_2():
    print('任务2...')

    time.sleep(2)


# 通过 Thread 类创建线程对象, 并使用 target 绑定线程对象要运行的任务
t1 = threading.Thread(target=work_1)
t2 = threading.Thread(target=work_2)

# 运行线程
start = time.time()

t1.start()
t2.start()

t1.join()
t2.join()

end = time.time()

print(f'总共用时: {end - start} s')

任务1...
任务2...
总共用时: 2.008962392807007 s

• 可以看到主线程等待子线程运行结束后才继续向下执行，正确记录了子线程运行的时间

thread_object.setDaemon()

• 设置守护线程，需要在线程启动之前进行设置
• 如果一个线程是守护线程，那么主线程运行到文件末尾后不论子线程任务是否结束都会自动退出

没有设置守护线程的情况

import time
import threading


def work():
    for i in range(5):
        print(i)

        time.sleep(1)


t = threading.Thread(target=work)
# t.setDaemon(True)

t.start()

print('主线程即将退出...')

0
主线程即将退出...
1
2
3
4

设置守护线程的情况

import time
import threading


def work():
    for i in range(5):
        print(i)

        time.sleep(1)


t = threading.Thread(target=work)
t.setDaemon(True)

t.start()

print('主线程即将退出...')

0
主线程即将退出...

• 可以看到并没有继续输出$1$、$2$、$3$、$4$，主线程就退出了

thread_object.current_thread()

• thread_object.current_thread()用于获取当前线程对象的引用
• 可以用来获取线程名称

import threading


def work():
    name = threading.current_thread().name  # .getName()

    print(name)


for i in range(5):
    t = threading.Thread(target=work)
    t.name = f'线程-{i}'  # setName(f'线程-{i}')

    t.start()

线程-0
线程-1
线程-2
线程-3
线程-4

---

目前爬虫的三种实现

单线程爬虫

import requests


def get_image(url):
    response = requests.get(url).content

    file_name = url.split('/')[-1]
    with open(file_name, 'wb') as f:
        f.write(response)

    print('下载完成...')


url_list = [
    'http://pic.bizhi360.com/bbpic/98/10798.jpg',
    'http://pic.bizhi360.com/bbpic/92/10792.jpg',
    'http://pic.bizhi360.com/bbpic/86/10386.jpg'
]

for img_url in url_list:
    get_image(img_url)

多线程并发爬虫

import requests
import threading


def get_image(url):
    response = requests.get(url).content

    file_name = url.split('/')[-1]
    with open(file_name, 'wb') as f:
        f.write(response)

    print('下载完成...')


url_list = [
    'http://pic.bizhi360.com/bbpic/98/10798.jpg',
    'http://pic.bizhi360.com/bbpic/92/10792.jpg',
    'http://pic.bizhi360.com/bbpic/86/10386.jpg'
]

for img_url in url_list:
    t = threading.Thread(target=get_image, args=(img_url,))

    t.start()

多进程并发爬虫

import requests
import multiprocessing


def get_image(url):
    response = requests.get(url).content

    file_name = url.split('/')[-1]
    with open(file_name, 'wb') as f:
        f.write(response)

        print('下载完成...')


url_list = [
    'http://pic.bizhi360.com/bbpic/98/10798.jpg',
    'http://pic.bizhi360.com/bbpic/92/10792.jpg',
    'http://pic.bizhi360.com/bbpic/86/10386.jpg'
]

if __name__ == '__main__':
    for img_url in url_list:
        p = multiprocessing.Process(target=get_image, args=(img_url,))

        p.start()

---

GIL锁

• 在Python中存在全局解释器锁，即GIL锁

• GIL是CPython解释器独有的，主要的功能是让一个进程在同一时刻只有一个线程被执行

• 例如在一个进程中创建了多个线程，在运行当前程序时，同一时刻只能有一个线程被执行，其他线程等待CPU调度，这种情况下无法利用多核CPU的优势

• 如果想要绕开GIL，那么可以使用多进程的方式，创建多个进程，使每个进程只有一个主线程，但是多进程消耗的资源比多线程的方式多

• 所以如果任务是I/O密集型任务，优先使用多线程方式，如果任务是计算密集型任务，优先使用多进程方式

---

线程安全

• 如果多个线程对同一个全局变量进行更新操作，会产生资源竞争问题，会导致数据不一致，计算结果出错

不可控的调度

import threading


def add_():
    global counter

    for i in range(1000000):
        counter += 1


def sub_():
    global counter

    for i in range(1000000):
        counter -= 1


counter = 0

t1 = threading.Thread(target=add_)
t2 = threading.Thread(target=sub_)

t1.start()
t2.start()

t1.join()
t2.join()

print(counter)

-12078

• 可以看到线程t1对num进行$1000000$次加$1$操作，线程t2对num进行$1000000$次减$1$操作，预期的结果输出$0$，而实际结果输出一个奇怪的数$?12078$，多次重复运行会发现结果不尽相同
• 这是由于加法和减法操作并不是一个“原子操作”，即在系统底层是使用多条汇编代码实现的，例如加$1$操作的汇编代码如下

mov	0x8049a1c, %eax
add $0x1, %eax
mov %eax, 0x8049a1c

• 这个例子假定变量counter指向内存地址0x8049a1c，先用mov指令，从内存地址处取出值，放入eax寄存器，然后给eax寄存器的值加$1$（0x1），最后eax的值被存回内存中相同的地址
• 如果操作系统调度线程t1运行到这段代码，它将counter的值（假设它这时是$0$）加载到它的寄存器eax中，因此线程$1$的eax = 0，然后它向寄存器加$1$，因此eax = 1，然后此时发生了时钟中断，操作系统将当前正在运行的线程（它的程序计数器、寄存器，包括eax等）的状态保存到线程t1的TCB（线程控制块）中
• 然后糟糕的事情发生了，线程t2被调度运行，并进入下面这段代码

mov	0x8049a1c, %eax
sub $0x1, %eax
mov %eax, 0x8049a1c

• 线程t2执行了第一条指令，获取内存中counter的值并将其放入其eax中（每个线程都有自己的专用寄存器），此时counter的值仍为$0$，因此线程t2的eax = 0，假设线程t2执行接下来的两条指令，将eax减$1$（因此eax = -1），然后将eax的内容保存到 counter（地址0x8049a1c）中，全局变量counter现在的值是$?1$

• 然后，又发生一次上下文切换，线程t1恢复运行，它已经执行过mov和add指令，现在准备执行最后一条mov指令，线程t1的eax = 1，最后的mov指令执行，将值保存到内存，counter被设置为$1$

• 简单来说，就是执行了一次加$1$操作，一次减$1$操作，而counter被加了$1$，而不是预期的counter = 0

• 系统对线程的调度是无法在代码层面进行控制的

锁

• 使用锁可以解决上述问题，锁的作用就是对加锁的代码执行原子操作，即要么代码全部运行，要么代码全不运行

示例

from threading import Thread, RLock

lock_obj = RLock()


def add_():
    global counter

    for i in range(1000000):
        lock_obj.acquire()  # 申请锁, 申请成功会让其他线程等待直到当前线程释放锁

        counter += 1

        lock_obj.release()  # 释放锁, 当锁被释放后其他等待的线程才能被正常执行


def sub_():
    global counter

    for i in range(1000000):
        lock_obj.acquire()  # 申请锁, 申请成功会让其他线程等待直到当前线程释放锁

        counter -= 1

        lock_obj.release()  # 释放锁, 当锁被释放后其他等待的线程才能被正常执行


counter = 0

t1 = Thread(target=add_)
t2 = Thread(target=sub_)

t1.start()
t2.start()

t1.join()
t2.join()

print(counter)

0

• 可以看到此时counter的最终结果为$0$

使用上下文管理器管理锁

• 在上述代码中，我们手动申请锁与释放锁，RLock()方法支持上下文管理协议，可以使用with语句帮助我们申请和释放锁

from threading import Thread, RLock

lock_obj = RLock()


def add_():
    global counter

    for i in range(1000000):
        with lock_obj:
            counter += 1


def sub_():
    global counter

    for i in range(1000000):
        with lock_obj:
            counter -= 1


counter = 0

t1 = Thread(target=add_)
t2 = Thread(target=sub_)

t1.start()
t2.start()

t1.join()
t2.join()

print(counter)

0

RLock和Lock

• 在线程中一般使用两种锁机制：RLock和Lock

Lock

• Lock是同步锁，不支持锁嵌套，一般很少使用

from threading import Thread, Lock

lock_obj = Lock()


def add_():
    global counter

    for i in range(1000000):
        lock_obj.acquire()  # 申请锁, 申请成功会让其他线程等待直到当前线程释放锁
        lock_obj.acquire()

        counter += 1

        lock_obj.release()  # 释放锁, 当锁被释放后其他等待的线程才能被正常执行
        lock_obj.release()


def sub_():
    global counter

    for i in range(1000000):
        lock_obj.acquire()  # 申请锁, 申请成功会让其他线程等待直到当前线程释放锁
        lock_obj.acquire()

        counter -= 1

        lock_obj.release()  # 释放锁, 当锁被释放后其他等待的线程才能被正常执行
        lock_obj.release()


counter = 0

t1 = Thread(target=add_)
t2 = Thread(target=sub_)

t1.start()
t2.start()

t1.join()
t2.join()

print(counter)

• 当嵌套使用同步锁Lock时，产生了死锁，程序会卡死

RLock

• RLock是递归锁，支持锁嵌套

from threading import Thread, RLock

lock_obj = RLock()


def add_():
    global counter

    for i in range(1000000):
        lock_obj.acquire()  # 申请锁, 申请成功会让其他线程等待直到当前线程释放锁
        lock_obj.acquire()

        counter += 1

        lock_obj.release()  # 释放锁, 当锁被释放后其他等待的线程才能被正常执行
        lock_obj.release()


def sub_():
    global counter

    for i in range(1000000):
        lock_obj.acquire()  # 申请锁, 申请成功会让其他线程等待直到当前线程释放锁
        lock_obj.acquire()

        counter -= 1

        lock_obj.release()  # 释放锁, 当锁被释放后其他等待的线程才能被正常执行
        lock_obj.release()


counter = 0

t1 = Thread(target=add_)
t2 = Thread(target=sub_)

t1.start()
t2.start()

t1.join()
t2.join()

print(counter)

0

内置类型底层锁机制

• 在多线程编程中，不是任何地方都需要加锁处理，在一些内置类型（例如列表）底层已经进行了加锁

import threading

data_list = list()


def list_add():
    for _ in range(100000):
        data_list.append(i)

    print(len(data_list))


for i in range(2):
    t = threading.Thread(target=list_add)

    t.start()

100000
200000

• 可以看到没有进行加锁，两个线程仍然正确向列表分别插入了$100000$个元素

死锁

• 在使用锁的过程中，发现如果嵌套使用同步锁，程序会卡死，实际上此时产生了死锁
• 死锁是由于资源竞争而造成的一种堵塞现象
• 线程$1$持有锁$1$而等待锁$2$的释放，线程$2$持有锁$2$而等待锁$1$的释放，此时产生了循环等待，产生了死锁

示例

import threading
import time

mutex_1 = threading.Lock()
mutex_2 = threading.Lock()


class MyThread_1(threading.Thread):
    def run(self):
        # 对 mutex_1 上锁
        mutex_1.acquire()

        # mutex_1 上锁后, 延时 1 秒, 等待线程 2 对 mutex_2 上锁
        print(self.name + '---mutex_1-up---')

        time.sleep(1)

        # 此时会堵塞, 因为 mutex_2 已经被线程 2 抢先上锁了
        mutex_2.acquire()

        print(self.name + '---mutex_2-up---')

        # 对 mutex_2 解锁
        mutex_2.release()
        # 对 mutex_1 解锁
        mutex_1.release()


class MyThread_2(threading.Thread):
    def run(self):
        # 对 mutex_2 上锁
        mutex_2.acquire()

        # mutex_2 上锁后, 延时 1 秒, 等待线程 1 对 mutex_1 上锁
        print(self.name + '---mutex_2-up---')

        time.sleep(1)

        # 此时会堵塞, 因为 mutex_1 已经被线程 1 抢先上锁了
        mutex_1.acquire()

        print(self.name + '---mutex_1-up---')

        # 对 mutex_1 解锁
        mutex_1.release()
        # 对 mutex_2 解锁
        mutex_2.release()


if __name__ == '__main__':
    t1 = MyThread_1()
    t2 = MyThread_2()

    t1.start()
    t2.start()

Thread-1---mutex_1-up---
Thread-2---mutex_2-up---

• 由于产生了死锁，程序会卡死，不会终止

---

线程池

• 线程对象的创建需要时间，在需要创建大量线程对象的时候会发生性能下降的情况
• 使用线程池会创建出一定数量的线程对象，并且线程在执行完任务后不会被解释器销毁，这样下一个任务可以重复使用之前创建的这些线程对象

线程池的创建

import time
from concurrent.futures import ThreadPoolExecutor


def get_html(page):
    print(f'Successfully obtained page {page}...')

    time.sleep(1)

    return page


# 创建线程池对象
executor = ThreadPoolExecutor(max_workers=2)

# 通过 submit 提交需要执行的函数到线程池中
task_1 = executor.submit(get_html, 1)
task_2 = executor.submit(get_html, 2)

Successfully obtained page 1...
Successfully obtained page 2...

done()

• done()方法用于判断某个任务是否完成

import time
from concurrent.futures import ThreadPoolExecutor


def get_html(page):
    print(f'Successfully obtained page {page}...')

    time.sleep(1)

    return page


# 创建线程池对象
executor = ThreadPoolExecutor(max_workers=2)

# 通过 submit 提交需要执行的函数到线程池中
task_1 = executor.submit(get_html, 1)
task_2 = executor.submit(get_html, 2)

# done() 方法用于判断某个任务是否完成
print(f'task_1 完成情况: {task_1.done()}')

Successfully obtained page 1...
Successfully obtained page 2...
task_1 完成情况: False

cancel()

• cancel()方法用于取消未运行的任务，已经运行的任务无法被取消

import time
from concurrent.futures import ThreadPoolExecutor


def get_html(page):
    print(f'Successfully obtained page {page}...')

    time.sleep(1)

    return page


# 创建线程池对象
executor = ThreadPoolExecutor(max_workers=1)

# 通过 submit 提交需要执行的函数到线程池中
task_1 = executor.submit(get_html, 1)
task_2 = executor.submit(get_html, 2)

# cancel() 方法用于取消未运行的任务, 已经运行的任务无法被取消
print(f'task_2 任务取消: {task_2.cancel()}')

Successfully obtained page 1...
task_2 任务取消: True

• 通过将max_workers的值修改为$1$，使得task_2未能运行时就被取消

result()

• submit()方法的返回值是一个future对象
• 通过对future对象调用result()方法可以获取任务的返回值

import time
from concurrent.futures import ThreadPoolExecutor


def get_html(page):
    print(f'Successfully obtained page {page}...')

    time.sleep(1)

    return page


# 创建线程池对象
executor = ThreadPoolExecutor(max_workers=2)

# 通过 submit 提交需要执行的函数到线程池中
task_1 = executor.submit(get_html, 1)
task_2 = executor.submit(get_html, 2)

# 通过对 future 对象调用 result() 方法获取任务的返回值
print(f'task_1 返回结果: {task_1.result()}')
print(f'task_2 返回结果: {task_2.result()}')

Successfully obtained page 1...
Successfully obtained page 2...
task_1 返回结果: 1
task_2 返回结果: 2

as_completed()

• as_completed()用于获取已经执行成功的任务的返回值

import time
from concurrent.futures import ThreadPoolExecutor, as_completed


def get_html(page):
    print(f'Successfully obtained page {page}...')

    time.sleep(1)

    return page


# 创建线程池对象
executor = ThreadPoolExecutor(max_workers=2)

page_list = [1, 2, 3, 4]

# 批量提交任务并获取已经执行成功的任务的返回值
all_tasks = [executor.submit(get_html, page) for page in page_list]

# 只要线程任务执行完就能获取到返回值, 完成一个任务获取一个任务的返回值
for future in as_completed(all_tasks):
    data = future.result()

    print(f'Get data {data}')

Successfully obtained page 1...
Successfully obtained page 2...
Successfully obtained page 3...
Successfully obtained page 4...
Get data 2
Get data 1
Get data 4
Get data 3

map()

• map()用于提交任务并获取已经执行成功的任务的返回值

import time
from concurrent.futures import ThreadPoolExecutor


def get_html(page):
    print(f'Successfully obtained page {page}...')

    time.sleep(1)

    return page


# 创建线程池对象
executor = ThreadPoolExecutor(max_workers=2)

page_list = [1, 2, 3, 4]

# map() 用于提交任务并获取已经执行成功的任务的返回值
for data in executor.map(get_html, page_list):
    print(f'Get data {data}')  # 打印的返回值顺序与列表顺序一致

Successfully obtained page 1...
Successfully obtained page 2...
Successfully obtained page 3...
Successfully obtained page 4...
Get data 1
Get data 2
Get data 3
Get data 4

wait()

• wait()方法用于使主线程堵塞，直到指定任务完成后，主线程才解堵塞

import time
from concurrent.futures import ThreadPoolExecutor, wait


def get_html(page):
    print(f'Successfully obtained page {page}...')

    time.sleep(1)

    return page


# 创建线程池对象
executor = ThreadPoolExecutor(max_workers=2)

page_list = [1, 2, 3, 4]

# 批量提交任务并获取已经执行成功的任务的返回值
all_tasks = [executor.submit(get_html, page) for page in page_list]

# wait() 方法用于使主线程堵塞, 直到指定任务完成后, 主线程才解堵塞
wait(all_tasks)

print('主线程解堵塞, 执行剩余代码')

Successfully obtained page 1...
Successfully obtained page 2...
Successfully obtained page 3...
Successfully obtained page 4...
主线程解堵塞, 执行剩余代码

---

自定义线程类

• 自定义线程类需要继承Thread类
• 需要重写Thread类中的run()方法，用于运行线程任务

示例

import requests
import threading


class ThreadSpider(threading.Thread):
    def __init__(self, url):
        super().__init__()

        self.url = url

    def run(self):
        response = requests.get(self.url).content

        file_name = self.url.split('/')[-1]
        with open(file_name, 'wb') as f:
            f.write(response)

        print('下载完成...')


url_list = [
    'http://pic.bizhi360.com/bbpic/98/10798.jpg',
    'http://pic.bizhi360.com/bbpic/92/10792.jpg',
    'http://pic.bizhi360.com/bbpic/86/10386.jpg'
]

for img_url in url_list:
    thread_spider = ThreadSpider(img_url)

    thread_spider.start()

下载完成...
下载完成...
下载完成...

---