devbox
2023面试高手
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Python3多线程详解_python3 多线程

作者：2023面试高手 | 2024-02-26 03:03:02

赞

踩

python3 多线程

为什么要使用多线程？

使用多线程，可以同时进行多项任务，可以使用户界面更友好，还可以后台执行某些用时长的任务，同时具有易于通信的优点。（对于GIL以及Python多线程对于效率的影响讨论可看知乎为什么有人说 Python 的多线程是鸡肋呢？ - 知乎 (zhihu.com)）

python3中多线程的实现使用了threading模块，它允许同一进程中运行多个线程。

如何创建和执行一个线程

一般我们有两种方法来创建线程，一种是以某个函数来作为起点，另一种是继承Thread类。

方法一

获取一个Thread对象，构造参数中target是起点函数，注意不要加括号。假如起点函数有参数，则可以通过args输入元组参数或者kwargs输入字典参数。

#! -*-conding=: UTF-8 -*-
# 2023/5/6 15:53
import time
from threading import Thread

def task():
    print("另外开始一个子线程做任务啦")
    time.sleep(1)  # 用time.sleep模拟任务耗时
    print("子线程任务结束啦")

if __name__ == '__main__':
    print("这里是主线程")
    # 创建线程对象
    t1 = Thread(target=task)
    # 启动
    t1.start()
    time.sleep(0.3)
    print("主线程依然可以干别的事")


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

输出结果为：

这里是主线程
另外开始一个子线程做任务啦
主线程依然可以干别的事
子线程任务结束啦


  • 1
  • 2
  • 3
  • 4
  • 5
方法二
#! -*-conding=: UTF-8 -*-
# 2023/5/6 15:53
import time
from threading import Thread

class NewThread(Thread):
    def __init__(self):
        Thread.__init__(self)  # 必须步骤

    def run(self):  # 入口是名字为run的方法
        print("开始新的线程做一个任务啦")
        time.sleep(1)  # 用time.sleep模拟任务耗时
        print("这个新线程中的任务结束啦")

if __name__ == '__main__':
    print("这里是主线程")
    # 创建线程对象
    t1 = NewThread()
    # 启动
    t1.start()
    time.sleep(0.3)  # 这里如果主线程结束，子线程会立刻退出，暂时先用sleep规避
    print("主线程依然可以干别的事")


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

输出结果为：

这里是主线程
开始新的线程做一个任务啦
主线程依然可以干别的事
这个新线程中的任务结束啦


  • 1
  • 2
  • 3
  • 4
  • 5

正式介绍threading模块

docs.python.org/3/library/t…

关于线程信息的函数：

  • threading.active_count()：返回当前存活的Thread对象数量。
  • threading.current_thread()：返回当前线程的Thread对象。
  • threading.enumerate()：列表形式返回所有存活的Thread对象。
  • threading.main_thread()：返回主Thread对象。

Thread对象的方法及属性：

  • Thread.name：线程的名字，没有语义，可以相同名称。
  • Thread.ident：线程标识符，非零整数。
  • Thread.Daemon：是否为守护线程。
  • Thread.is_alive()：是否存活。
  • Thread.start()：开始线程活动。若多次调用抛出RuntimeError。
  • Thread.run()：用来重载的，
  • Thread.join(timeout=None)：等待直到线程正常或异常结束。尚未开始抛出RuntimeError
  • Thread(group=None, target=None, name=None, args=(), kwargs={}, *, deamon=None)：构造函数。

守护线程 Daemon

在Python 3中，守护线程（daemon thread）是一种特殊的线程，它在程序运行时在后台运行，不会阻止程序的退出。当主线程退出时，守护线程也不会自动退出，而不需要等待它执行完毕。

方法一

在创建线程对象时，可以通过设置daemon属性为True来创建守护线程，例如：

import threading
import time

def worker():
    while True:
        print('Worker thread running')
        time.sleep(1)

# 创建守护线程
t = threading.Thread(target=worker, daemon=True)
# 启动线程
t.start()

# 主线程执行一些操作
print('Main thread running')
time.sleep(5)
print('Main thread finished')


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18

输出结果为：

Worker thread runningMain thread running

Worker thread running
Worker thread running
Worker thread running
Worker thread running
Main thread finished


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

在这个示例中，我们创建了一个守护线程worker()，并将daemon属性设置为True。在主线程中，我们执行了一些操作，并休眠5秒钟。由于守护线程的存在，即使主线程已经结束，守护线程仍会在后台运行。

方法二

设置守护线程用Thread.setDaemon(bool)

#! -*-conding=: UTF-8 -*-
# 2023/5/6 16:06

import time
from threading import Thread

def task1():
    print("开始子线程1做任务1啦")
    time.sleep(1)  # 用time.sleep模拟任务耗时
    print("子线程1中的任务1结束啦")

def task2():
    print("开始子线程2做任务2啦")
    for i in range(5):
        print("任务2-{}".format(i))
        time.sleep(1)
    print("子线程2中的任务2结束啦")

if __name__ == '__main__':
    print("这里是主线程")
    # 创建线程对象
    t1 = Thread(target=task1)
    t2 = Thread(target=task2)
    t2.setDaemon(True)  # 设置为守护进程，必须在start之前
    # 启动
    t1.start()
    t2.start()
    time.sleep(1)
    print("主线程结束了")


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30

输出结果为：

这里是主线程
开始子线程1做任务1啦
开始子线程2做任务2啦
任务2-0
主线程结束了
子线程1中的任务1结束啦任务2-1


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

守护线程的作用在于，当我们需要在程序运行时执行一些后台任务，但是不想让这些任务阻止程序的正常退出时，可以使用守护线程。
例如，在一个Web应用程序中，我们可能需要启动一个守护线程来定期清理缓存或者执行一些后台任务。

需要注意的是，守护线程无法完全控制其执行过程，因此不能用于一些必须在程序退出之前完成的任务。同时，守护线程不能访问一些主线程资源，例如共享内存或者打开的文件，因为这些资源可能会在主线程结束时被释放。

让主线程等待子线程结束 join

假如要让主线程等子线程结束，那么可以使用Thread.join()方法。

当调用线程对象的join()方法时，主线程将被阻塞，直到该线程执行完成或者超时。

以下是一个简单的示例：

import threading
import time

def worker():
    print('Worker thread started')
    time.sleep(2)
    print('Worker thread finished')

# 创建线程对象
t = threading.Thread(target=worker)
# 启动线程
t.start()

# 等待线程结束
t.join()

# 主线程继续执行
print('Main thread finished')


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

输出结果为：

Worker thread started
Worker thread finished
Main thread finished


  • 1
  • 2
  • 3
  • 4

在这个示例中，我们创建了一个子线程worker()，并使用start()方法启动线程。在主线程中，我们调用了线程对象的join()方法，让主线程等待子线程执行完毕。在子线程执行完毕后，主线程继续执行。

需要注意的是，join()方法还可以设置超时时间，以避免无限期等待线程的执行。例如：

import threading
import time

def worker():
    print('Worker thread started')
    time.sleep(2)
    print('Worker thread finished')

# 创建线程对象
t = threading.Thread(target=worker)
# 启动线程
t.start()

# 等待线程结束，最多等待3秒钟
t.join(3)

# 主线程继续执行
print('Main thread finished')


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

输出结果为：

Worker thread started
Worker thread finished
Main thread finished


  • 1
  • 2
  • 3
  • 4

在这个示例中，我们设置了join()方法的超时时间为3秒钟，即使子线程没有执行完成，主线程也会在3秒钟后继续执行。

线程共享资源可能引起什么问题？

在线程编程中，多个线程可能同时访问和修改同一个共享资源，例如全局变量、共享内存、文件等。如果没有进行适当的同步操作，就可能会引发以下问题：

竞态条件（Race Condition）：当多个线程同时访问和修改同一个共享资源时，就可能会发生竞态条件。这种情况下，由于线程执行顺序的不确定性，可能会导致资源被错误地读取或写入，从而引发程序的错误或崩溃。

死锁（Deadlock）：当多个线程都在等待另一个线程释放某个资源时，就可能会发生死锁。这种情况下，程序会永久地阻塞在这个状态下，无法继续执行。

活锁（Livelock）：多个线程相互协作，但是由于某些原因无法前进，导致它们不断重试，最终导致系统陷入死循环。活锁是一种比死锁更难以诊断和解决的问题。

为了避免以上问题，我们可以使用线程同步机制来保护共享资源的访问。

例如，可以使用锁（Lock）、信号量（Semaphore）、条件变量（Condition）等机制来限制同时访问共享资源的线程数量，从而避免竞态条件。同时，也可以使用一些算法和策略来避免死锁和活锁等问题的发生。

下面是一些具体的例子，说明在多线程程序中共享资源可能引发的问题：

竞态条件
from threading import Thread

num = 0

def sum_one(quantity):
    global num
    for index in range(quantity):
        num += 1
    return num


t1 = Thread(target=sum_one, args=(1000000,))
t2 = Thread(target=sum_one, args=(2000000,))
t3 = Thread(target=sum_one, args=(3000000,))
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()
print("num-------->", num)


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

输出结果为：

num--------> 1580567


  • 1
  • 2

在这个例子中，我们创建了3个线程来执行sum_one()函数，这个三个线程会分别对全局变量x都进行加1000000操作。由于多个线程同时访问和修改x变量，就会产生竞态条件，导致x的最终值可能小于3000000。

死锁
import threading

lock1 = threading.Lock()
lock2 = threading.Lock()

def worker1():
    print('Worker 1 acquiring lock 1')
    lock1.acquire()
    print('Worker 1 acquired lock 1')
    print('Worker 1 acquiring lock 2')
    lock2.acquire()
    print('Worker 1 acquired lock 2')
    lock2.release()
    lock1.release()

def worker2():
    print('Worker 2 acquiring lock 2')
    lock2.acquire()
    print('Worker 2 acquired lock 2')
    print('Worker 2 acquiring lock 1')
    lock1.acquire()
    print('Worker 2 acquired lock 1')
    lock1.release()
    lock2.release()

t1 = threading.Thread(target=worker1)
t2 = threading.Thread(target=worker2)
t1.start()
t2.start()
t1.join()
t2.join()

print('Main thread finished')


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34

在这个例子中，我们创建了两个线程worker1()和worker2()，它们都需要同时获取lock1和lock2两个锁来执行操作。由于worker1()先获取lock1，然后尝试获取lock2，而worker2()先获取lock2，然后尝试获取lock1，就可能会产生死锁的情况。

活锁
import threading

class Account:
    def __init__(self, balance):
        self.balance = balance
        self.lock = threading.Lock()

    def withdraw(self, amount):
        with self.lock:
            if self.balance < amount:
                print('Withdraw failed: not enough balance')
                return False
            print(f'Withdraw {amount} from account')
            self.balance -= amount
            return True

    def transfer(self, target, amount):
        while True:
            if self.withdraw(amount):
                if target.deposit(amount):
                    return True
                else:
                    self.deposit(amount)
            else:
                return False

    def deposit(self, amount):
        with self.lock:
            print(f'Deposit {amount} to account')
            self.balance += amount
            return True

def worker1(acc1, acc2):
    while True:
        acc1.transfer(acc2, 100)
        print('Worker 1: transfer complete')

def worker2(acc1, acc2):
    while True:
        acc2.transfer(acc1, 100)
        print('Worker 2: transfer complete')

acc1 = Account(1000)
acc2 = Account(1000)
t1 = threading.Thread(target=worker1, args=(acc1, acc2))
t2 = threading.Thread(target=worker2, args=(acc1, acc2))
t1.start()
t2.start()
t1.join()
t2.join()


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51

输出结果为:

Withdraw 100 from account
Deposit 100 to account
Worker 1: transfer complete
Withdraw 100 from account
Withdraw 100 from account
Deposit 100 to accountDeposit 100 to account
Worker 1: transfer complete

Worker 2: transfer completeWithdraw 100 from account
Deposit 100 to account
Worker 1: transfer complete
Withdraw 100 from account
Deposit 100 to account
Worker 1: transfer complete
Withdraw 100 from account
Deposit 100 to account

Worker 1: transfer completeWithdraw 100 from account
Deposit 100 to account
Worker 2: transfer complete
Withdraw 100 from account
Withdraw 100 from account

Deposit 100 to accountDeposit 100 to account
Worker 2: transfer complete

Worker 1: transfer completeWithdraw 100 from account
Deposit 100 to account
Worker 2: transfer complete
Withdraw 100 from account
Deposit 100 to account
Worker 2: transfer complete
Withdraw 100 from account
Deposit 100 to account

... 


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

在这个例子中，我们创建了两个账户acc1和acc2，并创建了两个线程worker1()和worker2()，它们不断地在这两个账户之间转账。

由于transfer()方法中需要获取锁来修改账户余额，但是两个线程的执行顺序可能会导致它们同时等待对方释放锁，从而无法前进，最终导致系统陷入活锁的状态。

具体来说，假设worker1()执行了acc1.transfer(acc2, 100)，然后进入了transfer()方法中的if self.withdraw(amount)分支，在等待acc1的锁。

此时，worker2()执行了acc2.transfer(acc1, 100)，然后也进入了transfer()方法中的if self.withdraw(amount)分支，在等待acc2的锁。由于acc1和acc2之间的转账是相互依赖的，因此这两个线程无法前进，会一直重试，最终导致系统陷入活锁的状态。

多线程的锁机制

在Python3中，锁机制是一种线程同步机制，它用于协调多个线程的并发访问共享资源，以避免竞态条件的发生。

Python 3中的多线程锁机制主要是通过threading模块中的Lock、RLock和Semaphore等类来实现的。

Lock类是最基本的锁，它提供了两个基本方法acquire()和release()，用于获取锁和释放锁。当一个线程调用acquire()方法时，如果该锁没有被其他线程获取，则该线程获取到该锁并进入临界区，否则该线程就会被阻塞，直到该锁被其他线程释放为止。

RLock类是可重入锁，它允许同一个线程多次获取该锁，每次获取都必须有对应的释放操作。如果一个线程已经获取到该锁，它可以再次获取该锁而不被阻塞，这就是可重入的特性。RLock类提供了acquire()和release()方法，与Lock类相同。

Semaphore类是信号量，它与锁类似，但可以允许多个线程同时访问某个资源，而不是像锁一样只允许一个线程访问。它提供了acquire()和release()方法，用于获取和释放资源。

下面是一个使用Lock类的示例代码：

import threading

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for i in range(100000):
        lock.acquire()
        counter += 1
        lock.release()

threads = []
for i in range(10):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(counter)


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

上面的代码中，我们定义了一个全局变量counter和一个Lock对象lock。increment()函数用于在循环中对counter进行100000次加1操作，而在每次加1之前，我们首先获取lock，加1操作完成之后再释放lock。这样保证了多个线程同时对counter进行操作时，不会产生竞争条件。

另外，还需要注意到，对于每个获取锁的线程，一定要记得在合适的地方释放锁，否则就会出现死锁的情况。

在多线程环境中，多个线程可能同时访问某个共享资源，这可能导致竞态条件的发生，从而导致程序出现不可预测的结果。为了避免这种情况的发生，我们可以使用锁机制来控制对共享资源的访问。在使用锁机制时，需要注意以下几点：

  1. 锁是一种互斥机制，即同一时刻只能有一个线程持有锁，其他线程必须等待该线程释放锁后才能继续执行。

  2. 在访问共享资源前，线程需要先获取锁。如果锁已经被其他线程持有，则线程会被阻塞，直到其他线程释放锁。

  3. 在访问共享资源后，线程需要释放锁，以便其他线程可以获取锁并访问共享资源。

  4. 在使用锁时，需要保证所有线程都使用同一个锁对象。

锁机制可以用于解决多线程程序中的竞态条件、死锁和活锁等问题。

下面我们分别通过例子来说明锁是如何解决这些问题的。

竞态条件

竞态条件指的是多个线程对共享资源的竞争，导致结果的正确性取决于线程的执行顺序。

比如，在一个多线程程序中，多个线程同时对同一个变量进行加减操作，结果可能取决于每个线程的执行顺序，这就是一个典型的竞态条件。

通过使用锁，可以保证在任何时刻只有一个线程能够访问共享资源，从而避免竞态条件的出现。下面的例子演示了如何使用锁来解决竞态条件：

import threading

class Counter:
    def __init__(self):
        self.count = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            self.count += 1

def worker(counter, num_iters):
    for i in range(num_iters):
        counter.increment()

counter = Counter()
num_threads = 10
num_iters = 10000
threads = [threading.Thread(target=worker, args=(counter, num_iters)) for _ in range(num_threads)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(counter.count)


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26

输出结果为：

10000


  • 1
  • 2

在这个例子中，多个线程对 Counter 对象的 count 属性进行加 1 操作，这可能会导致竞态条件。

为了避免这种情况，我们使用了一个锁，通过 with self.lock 来获取锁，这样在任何时刻只有一个线程能够修改 count 属性。

这样，我们就避免了竞态条件的出现。

死锁

死锁是指两个或多个线程在等待彼此释放资源，从而形成僵局的情况。为了解决死锁问题，可以使用锁机制来协调线程对共享资源的访问。具体来说，当一个线程获得锁时，其他线程必须等待该线程释放锁之后才能访问共享资源，从而避免多个线程同时访问同一个共享资源而产生死锁。

例如，考虑一个简单的场景，其中有两个线程，分别需要获取两个共享资源才能继续执行。假设这两个线程在获取资源时的顺序不同，可能会出现死锁的情况。

import threading

resource_a = threading.Lock()
resource_b = threading.Lock()

def thread_a():
    resource_a.acquire()
    resource_b.acquire()
    print("Thread A acquired resource A and resource B")
    resource_b.release()
    resource_a.release()

def thread_b():
    resource_b.acquire()
    resource_a.acquire()
    print("Thread B acquired resource A and resource B")
    resource_a.release()
    resource_b.release()

thread1 = threading.Thread(target=thread_a)
thread2 = threading.Thread(target=thread_b)

thread1.start()
thread2.start()

thread1.join()
thread2.join()


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28

上述代码中，thread_a和thread_b分别获取resource_a和resource_b，但是它们的获取顺序不同。因此，如果这两个线程同时运行，就有可能发生死锁的情况，导致程序卡住。

为了避免死锁，可以使用锁机制。修改上述代码，如下所示：

import threading

resource_a = threading.Lock()
resource_b = threading.Lock()

def thread_a():
    resource_a.acquire()
    resource_b.acquire()
    print("Thread A acquired resource A and resource B")
    resource_b.release()
    resource_a.release()

def thread_b():
    resource_a.acquire()
    resource_b.acquire()
    print("Thread B acquired resource A and resource B")
    resource_b.release()
    resource_a.release()

thread1 = threading.Thread(target=thread_a)
thread2 = threading.Thread(target=thread_b)

thread1.start()
thread2.start()

thread1.join()
thread2.join()


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28

在这个示例中，每个线程都按照相同的顺序获取锁，这样就避免了死锁的情况。

活锁

活锁是多线程程序中的一种常见问题，它是指线程在尝试协调其操作时一直重试，但最终没有达到进展的状态。一个常见的例子是两个线程互相等待对方释放其持有的资源。

使用锁是解决活锁问题的一种常见方式。当线程需要访问共享资源时，必须获得相应的锁。如果锁已经被其他线程持有，线程将阻塞直到获得锁为止。这样，当多个线程尝试同时访问同一共享资源时，只有一个线程能够获取锁，其他线程将被阻塞。

下面是一个使用锁解决活锁问题的例子。假设有两个线程A和B，它们需要同时访问两个共享资源x和y，但由于资源x和y的访问顺序不同，线程A需要先获得x再获得y，而线程B需要先获得y再获得x。如果两个线程尝试同时获取它们需要的资源，就会出现活锁问题。

使用锁可以解决这个问题。假设每个线程都先获取x的锁，然后再获取y的锁，这样就可以保证每个线程都按照相同的顺序获取资源，避免了死锁和活锁的问题。

下面是使用锁解决活锁问题的代码示例：

import threading

class Resource:
    def __init__(self):
        self.lock1 = threading.Lock()
        self.lock2 = threading.Lock()

    def get_x(self):
        self.lock1.acquire()
        return "x"

    def get_y(self):
        self.lock2.acquire()
        return "y"

    def release_x(self):
        self.lock1.release()

    def release_y(self):
        self.lock2.release()

def thread_a(resource):
    while True:
        x = resource.get_x()
        y = resource.get_y()
        print("Thread A got resources x and y")
        resource.release_x()
        resource.release_y()

def thread_b(resource):
    while True:
        y = resource.get_y()
        x = resource.get_x()
        print("Thread B got resources y and x")
        resource.release_y()
        resource.release_x()

if __name__ == "__main__":
    resource = Resource()
    a = threading.Thread(target=thread_a, args=(resource,))
    b = threading.Thread(target=thread_b, args=(resource,))
    a.start()
    b.start()


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44

在这个例子中，每个线程都使用相同的锁顺序来获取资源x和y，这样就避免了活锁的问题。

使用锁可能导致执行速度慢，但是保证了线程安全
无论是Lock还是RLock，acquire和release都要成对出现

多线程的通信

Python3 中多线程之间的通信方式有以下几种：

队列

在 Python 3 中，可以使用队列（Queue）实现多线程之间的通信。队列是线程安全的数据结构，可以实现线程之间的同步和协作，避免竞争条件和死锁问题。

Python 内置了 Queue 模块，提供了队列数据结构，它可以用于实现多线程之间的安全通信。可以使用队列的 put() 方法往队列中添加元素，使用 get() 方法从队列中取出元素。

Queue模块提供了以下几个类：

  • Queue：基本队列，实现FIFO（先进先出）的算法。
  • LifoQueue：与Queue类似，但是实现了LIFO（后进先出）的算法。
  • PriorityQueue：队列中的每个元素都有一个优先级，每次弹出优先级最高的元素。
  • SimpleQueue：类似于Queue，但是没有任务协作的功能，也就是说，不能在进程之间使用。

Queue类中最常用的方法包括：

  • put(item[, block[, timeout]])：将一个item放入队列，如果队列已满，block为True则阻塞，直到队列未满或超时；block为False时，则抛出queue.Full异常。
  • get([block[, timeout]])：从队列中取出并返回一个item，如果队列为空，block为True则阻塞，直到队列不为空或超时；block为False时，则抛出queue.Empty异常。
  • task_done()：通知队列，一个先前放入队列的任务已经完成。
  • join()：阻塞主线程，直到队列中所有的任务都被处理完。

下面举一个简单的例子：

import threading
import queue
import time

# 生产者线程，负责向队列中添加数据
class ProducerThread(threading.Thread):
    def __init__(self, queue, name):
        threading.Thread.__init__(self)
        self.queue = queue
        self.name = name

    def run(self):
        for i in range(5):
            item = "item-" + str(i)
            self.queue.put(item)
            print(self.name, "produced", item)
            time.sleep(1)

# 消费者线程，负责从队列中取出数据
class ConsumerThread(threading.Thread):
    def __init__(self, queue, name):
        threading.Thread.__init__(self)
        self.queue = queue
        self.name = name

    def run(self):
        while True:
            item = self.queue.get()
            if item is None:
                break
            print(self.name, "consumed", item)
            time.sleep(0.5)

# 创建一个队列对象
q = queue.Queue()

# 创建两个线程对象，分别作为生产者和消费者
producer = ProducerThread(q, "Producer")
consumer = ConsumerThread(q, "Consumer")

# 启动线程
producer.start()
consumer.start()

# 等待生产者线程完成生产
producer.join()

# 停止消费者线程
q.put(None)
consumer.join()


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51

在上面的例子中，我们创建了一个生产者线程和一个消费者线程。生产者线程负责向队列中添加数据，消费者线程负责从队列中取出数据。生产者线程每隔一秒钟向队列中添加一个字符串，消费者线程每隔半秒钟从队列中取出一个字符串。为了避免消费者线程在队列为空时陷入死循环，我们在队列的末尾放置了一个 None 值，当消费者线程取出该值时，就会退出循环。

事件（Event）

事件是一种同步对象，可以用于多线程之间的通信，常用于控制线程的执行顺序。可以使用事件的 set() 方法设置事件，使用 wait() 方法等待事件被设置，使用 clear() 方法清除事件。

以下是一个使用事件实现多线程间通信的示例代码：

import time
import threading

def worker1(event):
    print('Worker 1 is waiting...\n')
    event.wait()
    print('Worker 1 is running...\n')

def worker2(event):
    print('Worker 2 is waiting...\n')
    event.wait()
    print('Worker 2 is running...\n')


event = threading.Event()

t1 = threading.Thread(target=worker1, args=(event,))
t2 = threading.Thread(target=worker2, args=(event,))

t1.start()
t2.start()

print('Main thread is sleeping...\n')
time.sleep(3)
event.set()

t1.join()
t2.join()


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

输出结果为：

Worker 1 is waiting...
Worker 2 is waiting...
Main thread is sleeping...

Worker 1 is running...
Worker 2 is running...


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

该代码创建了两个线程，它们都等待事件被设置，当事件被设置后，它们才开始执行。在主线程中，先休眠了 3 秒钟，然后设置了事件，从而唤醒了两个线程。在实际应用中，事件可以用于控制线程的执行顺序，或者实现线程之间的协作。

锁（Lock）

使用锁可以实现多线程间的通信，可以通过共享变量和锁的机制来实现线程间的同步和互斥。具体来说，一个线程需要访问共享变量时，首先需要获得锁，然后读取或修改共享变量的值，完成操作后再释放锁，以便其他线程访问共享变量。

  • 下面是一个简单的示例代码，其中两个线程共享一个变量 counter，通过锁的机制来实现对该变量的互斥访问。
import threading

class CounterThread(threading.Thread):
    def __init__(self, lock):
        super().__init__()
        self.lock = lock

    def run(self):
        global counter
        for i in range(100000):
            with self.lock:
                counter += 1

if __name__ == '__main__':
    lock = threading.Lock()
    counter = 0
    threads = [CounterThread(lock) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f'counter = {counter}')


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23

输出结果为：

counter = 1000000


  • 1
  • 2

在这个例子中，CounterThread 是一个继承自 threading.Thread 的线程类，它有一个成员变量 lock，用于控制对共享变量 counter 的访问。在 run 方法中，线程会循环执行一定次数的加操作，每次操作前会获取锁并对 counter 做加一操作，完成后再释放锁。在主线程中创建了 10 个 CounterThread 线程，并启动它们进行计数操作。在所有线程都执行完毕后，打印出 counter 的最终值。

使用锁可以确保多个线程对共享变量的访问是互斥的，从而避免竞态条件和数据损坏等问题。但是，使用锁也可能会导致性能问题和死锁等问题，因此需要谨慎使用，并根据实际情况选择合适的同步机制。

  • 或者
import threading

class Counter:
    def __init__(self):
        self.value = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            self.value += 1

def worker(counter, n):
    for i in range(n):
        counter.increment()

counter = Counter()
threads = []
for i in range(10):
    t = threading.Thread(target=worker, args=(counter, 10000))
    threads.append(t)

for t in threads:
    t.start()

for t in threads:
    t.join()

print(counter.value)


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

在这个示例中，我们创建了一个 Counter 类，其中包含一个整数 value 和一个锁对象 lock。 increment 方法使用 with 语句获取锁并增加 value 的值。

我们还创建了 10 个线程，每个线程都会调用 worker 函数。这个函数会循环 10000 次调用 increment 方法来增加 value 的值。

由于每个线程都会获取锁来访问共享资源，因此只有一个线程可以访问 increment 方法，避免了多个线程同时修改 value 的值，从而确保了线程安全。最终的输出结果应该是100000，即10个线程分别增加了10000次。

条件变量（Condition）实现多线程间的通信

条件变量（Condition）是Python多线程编程中常用的线程间通信机制之一，它可以用于线程间的通信和同步，提供了一个线程等待另一个线程通知其发生了某个事件的方法。

下面是一个使用条件变量实现多线程间通信的示例代码：

import threading
import time

class Producer(threading.Thread):
    def __init__(self, name, cond):
        super().__init__(name=name)
        self.cond = cond

    def run(self):
        for i in range(10):
            with self.cond:
                print(f"{self.name} producing item {i}")
                self.cond.notify() # 通知消费者线程
                self.cond.wait() # 等待消费者线程通知

class Consumer(threading.Thread):
    def __init__(self, name, cond):
        super().__init__(name=name)
        self.cond = cond

    def run(self):
        for i in range(10):
            with self.cond:
                self.cond.wait() # 等待生产者线程通知
                print(f"{self.name} consuming item {i}")
                self.cond.notify() # 通知生产者线程

cond = threading.Condition()
producer = Producer("Producer", cond)
consumer = Consumer("Consumer", cond)
producer.start()
consumer.start()
producer.join()
consumer.join()


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35

在这个示例代码中，生产者线程通过cond.notify()方法通知消费者线程，消费者线程通过cond.wait()方法等待生产者线程的通知。条件变量cond用于实现线程之间的同步和通信，生产者线程和消费者线程在共享同一把锁的情况下，通过with self.cond:语句获取条件变量的锁并进入临界区，确保线程安全。

信号量（Semaphore）实现多线程间的通信

信号量（Semaphore）是一种用于控制并发访问共享资源的同步原语。它是一种计数器，用于控制多个线程对共享资源的访问。信号量维护一个计数器，初始值为一个非负整数，每当一个线程访问共享资源时，计数器减1；当计数器为0时，所有试图访问共享资源的线程都会被阻塞，直到某个线程释放了共享资源，此时计数器加1，被阻塞的线程才有机会继续执行。

在 Python 中，我们可以使用 threading.Semaphore 类来创建信号量对象。该类的构造函数接受一个整数作为参数，表示初始计数器的值。Semaphore 类有两个方法，acquire() 和 release()，分别用于获取和释放信号量。

以下是使用信号量实现的简单示例代码：

import threading

class Producer(threading.Thread):
    def __init__(self, name, buf, sem):
        super().__init__(name=name)
        self.buf = buf
        self.sem = sem

    def run(self):
        for i in range(5):
            self.sem.acquire()
            self.buf.append(i)
            print(f"{self.name} produced {i}")
            self.sem.release()

class Consumer(threading.Thread):
    def __init__(self, name, buf, sem):
        super().__init__(name=name)
        self.buf = buf
        self.sem = sem

    def run(self):
        while True:
            self.sem.acquire()
            if not self.buf:
                self.sem.release()
                break
            item = self.buf.pop(0)
            print(f"{self.name} consumed {item}")
            self.sem.release()

if __name__ == '__main__':
    buf = []
    sem = threading.Semaphore(1)
    producer = Producer("Producer", buf, sem)
    consumer1 = Consumer("Consumer1", buf, sem)
    consumer2 = Consumer("Consumer2", buf, sem)
    producer.start()
    consumer1.start()
    consumer2.start()
    producer.join()
    consumer1.join()
    consumer2.join()


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44

在这个示例代码中，有一个生产者线程和两个消费者线程。生产者线程向共享缓冲区中添加数据，而消费者线程从缓冲区中获取数据。为了避免竞争条件，我们使用了信号量。

在生产者线程中，当信号量可用时，它会获取信号量并添加数据到缓冲区中，然后释放信号量。在消费者线程中，当信号量可用时，它会获取信号量并从缓冲区中获取数据，然后释放信号量。

通过使用信号量，我们可以确保生产者和消费者线程之间的同步，从而避免了竞争条件和死锁问题。

管道（Pipe）

在 Python3 中，可以使用 multiprocessing 模块中的 Pipe 类来实现多进程间的通信，也可以用 multiprocessing.connection 模块来创建多进程间的通信通道。下面的例子是用 Pipe 类来实现多线程间的通信：

import threading
from multiprocessing import Pipe

def producer(pipe):
    for i in range(5):
        pipe.send(i)
        print(f"{threading.current_thread().name} produced {i}")
    pipe.close()

def consumer(pipe):
    while True:
        try:
            item = pipe.recv()
            print(f"{threading.current_thread().name} consumed {item}")
        except EOFError:
            break

if __name__ == '__main__':
    producer_conn, consumer_conn = Pipe()
    producer_thread = threading.Thread(target=producer, args=(producer_conn,))
    consumer_thread = threading.Thread(target=consumer, args=(consumer_conn,))
    producer_thread.start()
    consumer_thread.start()
    producer_thread.join()
    consumer_thread.join()


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26

在这个例子中，我们创建了两个线程，一个生产者线程和一个消费者线程。它们之间共享一个管道（Pipe），其中生产者将数据写入管道，而消费者从管道中读取数据。当生产者完成其工作后，它会关闭管道以通知消费者停止运行。

需要注意的是，在 Pipe 中，发送和接收操作是阻塞的。因此，在发送或接收数据时，如果没有可用的空间或数据，线程将被阻塞，直到有数据可用或空间可用。

定时器Timer

定时器（Timer）是Python中的一个线程类，它可以在一定时间之后调用指定的函数或方法。Timer是继承自Thread类的，因此可以像Thread一样启动、停止和等待它。

定时器的构造函数如下：

class threading.Timer(interval, function, args=None, kwargs=None)


  • 1
  • 2

其中，interval表示定时器的时间间隔（秒），function表示定时器超时后要调用的函数或方法。args和kwargs是传递给函数或方法的参数。

下面是一个使用定时器的例子：

import threading

def hello():
    print("Hello, world!")

t = threading.Timer(5.0, hello)
t.start()   # 启动定时器，等待5秒后调用hello函数


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

这个程序中，我们创建了一个定时器t，它会在5秒后调用hello函数，并启动定时器。程序将在启动定时器后立即继续执行，而定时器则在后台等待5秒，然后调用hello函数。

如果我们想要停止定时器，可以使用cancel()方法：

t.cancel()  # 停止定时器


  • 1
  • 2

需要注意的是，如果定时器已经超时并且在调用函数之前被取消，那么函数将不会被调用。因此，需要在调用cancel()方法之前等待定时器超时。

python3线程池

concurrent.futures实现多线程

Python 3中的线程池是一种常见的多线程编程模型，可以提高多线程程序的性能和可维护性。在Python 3中，线程池可以通过标准库中的concurrent.futures模块来实现。

concurrent.futures模块定义了两个类：ThreadPoolExecutor和ProcessPoolExecutor。这两个类都实现了Python 3中的执行器(Executor)接口，提供了一种方便的方式来异步执行函数或方法，并返回其结果。

Exectuor 提供了如下常用方法：

  • submit(fn, *args, **kwargs)：将 fn 函数提交给线程池。*args 代表传给 fn 函数的参数，*kwargs 代表以关键字参数的形式为 fn 函数传入参数。
  • map(func, *iterables, timeout=None, chunksize=1)：该函数类似于全局函数 map(func, *iterables)，只是该函数将会启动多个线程，以异步方式立即对 iterables 执行 map 处理。超时抛出TimeoutError错误。返回每个函数的结果，注意不是返回future。
  • shutdown(wait=True)：关闭线程池。关闭之后线程池不再接受新任务，但会将之前提交的任务完成。

程序将task函数submit给线程池后，会返回一个Future对象，Future主要用来获取task的返回值。

Future 提供了如下方法：

  • cancel()：取消该 Future 代表的线程任务。如果该任务正在执行，不可取消，则该方法返回 False；否则，程序会取消该任务，并返回 True。
  • cancelled()：返回 Future 代表的线程任务是否被成功取消。
  • running()：如果该 Future 代表的线程任务正在执行、不可被取消，该方法返回 True。
  • done()：如果该 Funture 代表的线程任务被成功取消或执行完成，则该方法返回 True。
  • result(timeout=None)：获取该 Future 代表的线程任务最后返回的结果。如果 Future 代表的线程任务还未完成，该方法将会阻塞当前线程，其中 timeout 参数指定最多阻塞多少秒。超时抛出TimeoutError，取消抛出CancelledError。
  • exception(timeout=None)：获取该 Future 代表的线程任务所引发的异常。如果该任务成功完成，没有异常，则该方法返回 None。
  • add_done_callback(fn)：为该 Future 代表的线程任务注册一个“回调函数”，当该任务成功完成时，程序会自动触发该 fn 函数，参数是future。

使用线程池来执行线程任务的步骤如下：

  1. 调用 ThreadPoolExecutor 类的构造器创建一个线程池。
  2. 定义一个普通函数作为线程任务。
  3. 调用 ThreadPoolExecutor 对象的 submit() 方法来提交线程任务。
  4. 当不想提交任何任务时，调用 ThreadPoolExecutor 对象的 shutdown() 方法来关闭线程池。

ThreadPoolExecutor是一个线程池执行器，可以用来执行异步任务，它管理着一个线程池，其中包含若干个线程。当一个任务被提交给执行器时，执行器会将其分配给一个线程来执行。当线程池中的所有线程都在执行任务时，新提交的任务会被放入队列中，直到有可用的线程为止。

以下是一个使用ThreadPoolExecutor的简单示例：

from concurrent.futures import ThreadPoolExecutor
import time

def worker(num):
    print(f"Worker {num} starting")
    time.sleep(1)
    print(f"Worker {num} finished")

if __name__ == '__main__':
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = [executor.submit(worker, i) for i in range(5)]
    for future in concurrent.futures.as_completed(futures):
        try:
            result = future.result()
        except Exception as e:
            print(f"Task raised an exception: {e}")
        else: 
            print(f"Task returned: {result}")


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19

在这个例子中，我们创建了一个线程池执行器，并指定了最大线程数为3。然后我们循环提交5个任务给执行器，每个任务都是一个worker函数，并传入不同的参数。由于我们设置了最大线程数为3，所以只会有3个任务同时被执行，另外2个任务会在之后的某个时间点被执行。

执行结果会存储在Future对象中，我们可以通过as_completed()方法获取任务的执行结果。如果任务执行过程中发生了异常，result()方法会抛出相应的异常。否则，它会返回任务的执行结果。

ThreadPoolExecutor还有其他一些有用的方法，例如shutdown()方法可以等待所有已提交的任务执行完毕并关闭线程池。

总之，Python 3中的线程池提供了一种方便的方式来执行异步任务，可以大大提高多线程程序的性能和可维护性。

使用线程池的好处和场景

使用线程池的优点是可以避免线程的频繁创建和销毁，从而提高线程的利用率，减少系统的开销。因此，当需要频繁执行短时间的任务时，可以考虑使用线程池。例如：

  1. 网络服务器：在服务器端接收客户端请求后，可以使用线程池来处理客户端请求，以提高服务器的并发性能。
  2. 图像处理：在图像处理过程中，需要频繁启动和停止线程来处理每个像素点的计算，使用线程池可以减少线程的创建和销毁，提高处理效率。
  3. 数据库连接池：在数据库操作中，需要频繁创建和销毁数据库连接，使用线程池可以减少这种开销，提高数据库操作的效率。

总之，当需要频繁执行短时间的任务时，可以考虑使用线程池来优化程序性能。

---------------------------END---------------------------

题外话

当下这个大数据时代不掌握一门编程语言怎么跟的上脚本呢？当下最火的编程语言Python前景一片光明！如果你也想跟上时代提升自己那么请看一下.

在这里插入图片描述

感兴趣的小伙伴，赠送全套Python学习资料，包含面试题、简历资料等具体看下方。


声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/2023面试高手/article/detail/143492
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。