一文搞懂 Python 多线程threading

在Python3中，通过threading模块提供线程的功能。原来的thread模块已废弃。但是threading模块中有个Thread类（大写的T，类名），是模块中最主要的线程类，一定要分清楚了，千万不要搞混了。

threading模块提供了一些比较实用的方法或者属性，例如：

方法与属性	描述
current_thread()	返回当前线程
active_count()	返回当前活跃的线程数，1个主线程+n个子线程
get_ident()	返回当前线程
enumerate()	返回当前活动 Thread 对象列表
main_thread()	返回主 Thread 对象
settrace(func)	为所有线程设置一个 trace 函数
setprofile(func)	为所有线程设置一个 profile 函数
stack_size([size])	返回新创建线程栈大小；或为后续创建的线程设定栈大小为 size
TIMEOUT_MAX	Lock.acquire(), RLock.acquire(), Condition.wait() 允许的最大超时时间

threading模块包含下面的类：

• Thread：基本线程类
• Lock：互斥锁
• RLock：可重入锁，使单一进程再次获得已持有的锁(递归锁)
• Condition：条件锁，使得一个线程等待另一个线程满足特定条件，比如改变状态或某个值。
• Semaphore：信号锁。为线程间共享的有限资源提供一个”计数器”，如果没有可用资源则会被阻塞。
• Event：事件锁，任意数量的线程等待某个事件的发生，在该事件发生后所有线程被激活
• Timer：一种计时器
• Barrier：Python3.2新增的“阻碍”类，必须达到指定数量的线程后才可以继续执行。

1. 多线程

有两种方式来创建线程：一种是继承Thread类，并重写它的run()方法；另一种是在实例化threading.Thread对象的时候，将线程要执行的任务函数作为参数传入线程。

第一种方法：

import threading

class MyThread(threading.Thread):
    def __init__(self, thread_name):
        # 注意：一定要显式的调用父类的初始化函数。
        super(MyThread, self).__init__(name=thread_name)

    def run(self):
        print("%s正在运行中......" % self.name)

if __name__ == '__main__':    
    for i in range(10):
        MyThread("thread-" + str(i)).start()

第二种方法：

import threading
import time

def show(arg):
    time.sleep(1)
    print('thread '+str(arg)+" running....")

if __name__ == '__main__':
    for i in range(10):
        t = threading.Thread(target=show, args=(i,))
        t.start()

对于Thread类，它的定义如下：

threading.Thread(self, group=None, target=None, name=None,
     args=(), kwargs=None, *, daemon=None)

• 参数group是预留的，用于将来扩展；
• 参数target是一个可调用对象，在线程启动后执行；
• 参数name是线程的名字。默认值为“Thread-N“，N是一个数字。
• 参数args和kwargs分别表示调用target时的参数列表和关键字参数。

Thread类定义了以下常用方法与属性：

方法与属性	说明
start()	启动线程，等待CPU调度
run()	线程被cpu调度后自动执行的方法
getName()、setName()和name	用于获取和设置线程的名称。
setDaemon()	设置为后台线程或前台线程（默认是False，前台线程）。如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止。如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程执行完成后，程序才停止。
ident	获取线程的标识符。线程标识符是一个非零整数，只有在调用了start()方法之后该属性才有效，否则它只返回None。
is_alive()	判断线程是否是激活的（alive）。从调用start()方法启动线程，到run()方法执行完毕或遇到未处理异常而中断这段时间内，线程是激活的。
isDaemon()方法和daemon属性	是否为守护线程
join([timeout])	调用该方法将会使主调线程堵塞，直到被调用线程运行结束或超时。参数timeout是一个数值类型，表示超时时间，如果未提供该参数，那么主调线程将一直堵塞到被调线程结束。

在多线程执行过程中，有一个特点要注意，那就是每个线程各执行各的任务，不等待其它的线程，自顾自的完成自己的任务，比如下面的例子：

import time
import threading

def doWaiting():
    print('start waiting:', time.strftime('%H:%M:%S'))
    time.sleep(3)
    print('stop waiting', time.strftime('%H:%M:%S'))

t = threading.Thread(target=doWaiting)
t.start()
# 确保线程t已经启动
time.sleep(1)
print('start job')
print('end job')

执行结果是：

start waiting: 10:50:35
start job
end job
stop waiting 10:50:38

Python默认会等待最后一个线程执行完毕后才退出。上面例子中，主线程没有等待子线程t执行完毕，而是啥都不管，继续往下执行它自己的代码，执行完毕后也没有结束整个程序，而是等待子线程t执行完毕，整个程序才结束。

有时候我们希望主线程等等子线程，不要“埋头往前跑”。那要怎么办？使用join()方法！如下所示：

import time
import threading

def doWaiting():
    print('start waiting:', time.strftime('%H:%M:%S'))
    time.sleep(3)
    print('stop waiting', time.strftime('%H:%M:%S'))

t = threading.Thread(target=doWaiting)
t.start()
# 确保线程t已经启动
time.sleep(1)
print('start join')
# 将一直堵塞，直到t运行结束。
t.join()
print('end join')

执行结果：

start waiting: 10:54:03
start join
stop waiting 10:54:06
end join

我们还可以使用setDaemon(True)把所有的子线程都变成主线程的守护线程，当主线程结束后，守护子线程也会随之结束，整个程序也跟着退出。

import threading
import time

def run():
    print(threading.current_thread().getName(), "开始工作")
    time.sleep(2)       # 子线程停2s
    print("子线程工作完毕")

for i in range(3):
    t = threading.Thread(target=run,)
    t.setDaemon(True)   # 把子线程设置为守护线程，必须在start()之前设置
    t.start()

time.sleep(1)     # 主线程停1秒
print("主线程结束了！")
print(threading.active_count())  # 输出活跃的线程数

执行结果：

Thread-1 开始工作
Thread-2 开始工作
Thread-3 开始工作
主线程结束了！
4

2. 自定义线程类

对于threading模块中的Thread类，本质上是执行了它的run方法。因此可以自定义线程类，让它继承Thread类，然后重写run方法。

import threading

class MyThreading(threading.Thread):

    def __init__(self, func, arg):
        super(MyThreading,self).__init__()
        self.func = func
        self.arg = arg

    def run(self):
        self.func(self.arg)

def my_func(args):
    """
    你可以把任何你想让线程做的事定义在这里
    """
    pass

obj = MyThreading(my_func, 123)
obj.start()

3.线程锁

由于线程之间的任务执行是CPU进行随机调度的，并且每个线程可能只执行了n条指令之后就被切换到别的线程了。当多个线程同时操作一个对象，如果没有很好地保护该对象，会造成程序结果的不可预期，这被称为“线程不安全”。为了保证数据安全，我们设计了线程锁，即同一时刻只允许一个线程操作该数据。线程锁用于锁定资源，可以同时使用多个锁，当你需要独占某一资源时，任何一个锁都可以锁这个资源，就好比你用不同的锁都可以把相同的一个箱子锁住是一个道理。

我们先看一下没有锁的情况下，脏数据是如何产生的。

import threading
import time

number = 0

def plus():
    global number       # global声明此处的number是外面的全局变量number
    for _ in range(1000000):    # 进行一个大数级别的循环加一运算
        number += 1
    print("子线程%s运算结束后，number = %s" % (threading.current_thread().getName(), number))

for i in range(2):      # 用2个子线程，就可以观察到脏数据
    t = threading.Thread(target=plus)
    t.start()


time.sleep(2)       # 等待2秒，确保2个子线程都已经结束运算。
print("主线程执行完毕后，number = ", number)

执行结果（每次数值可能都不一样）：

子线程Thread-2运算结束后，number = 1144974
子线程Thread-1运算结束后，number = 1181608
主线程执行完毕后，number =  1181608

结果并不等于2,000,000，可以很明显地看出脏数据的情况。这是因为两个线程在运行过程中，CPU随机调度，你算一会我算一会，在没有对number进行保护的情况下，就发生了数据错误。如果想获得正确结果，可以使用join()方法，让多线程变成顺序执行，如下修改代码片段：

for i in range(2):     
    t = threading.Thread(target=plus)
    t.start()
    t.join()        # 添加这一行就让两个子线程变成了顺序执行

上面为了防止脏数据而使用join()的方法，其实是让多线程变成了单线程，属于因噎废食的做法，正确的做法是使用线程锁。Python在threading模块中定义了几种线程锁类，分别是：

• Lock 互斥锁
• RLock 可重入锁
• Semaphore 信号
• Event 事件
• Condition 条件
• Barrier “阻碍”

3.1 互斥锁Lock

互斥锁是一种独占锁，同一时刻只有一个线程可以访问共享的数据。使用很简单，初始化锁对象，然后将锁当做参数传递给任务函数，在任务中加锁，使用后释放锁。

import threading
import time

number = 0
lock = threading.Lock()

def plus(lk):
    global number       # global声明此处的number是外面的全局变量number
    lk.acquire()        # 开始加锁
    for _ in range(1000000):    # 进行一个大数级别的循环加一运算
        number += 1
    print("子线程%s运算结束后，number = %s" % (threading.current_thread().getName(), number))
    lk.release()        # 释放锁，让别的线程也可以访问number

if __name__ == '__main__':
    for i in range(2):      # 用2个子线程，就可以观察到脏数据
        t = threading.Thread(target=plus, args=(lock,)) # 需要把锁当做参数传递给plus函数
        t.start()
    time.sleep(2)       # 等待2秒，确保2个子线程都已经结束运算。
    print("主线程执行完毕后，number = ", number)

RLock的使用方法和Lock一模一样，只不过它支持重入锁。该锁对象内部维护着一个Lock和一个counter对象。counter对象记录了acquire的次数，使得资源可以被多次require。最后，当所有RLock被release后，其他线程才能获取资源。在同一个线程中，RLock.acquire()可以被多次调用，利用该特性，可以解决部分死锁问题。

3.2 信号Semaphore

类名：BoundedSemaphore。这种锁允许一定数量的线程同时更改数据，它不是互斥锁。比如地铁安检，排队人很多，工作人员只允许一定数量的人进入安检区，其它的人继续排队。

import time
import threading

def run(n, se):
    se.acquire()
    print("run the thread: %s" % n)
    time.sleep(1)
    se.release()

# 设置允许5个线程同时运行
semaphore = threading.BoundedSemaphore(5)
for i in range(20):
    t = threading.Thread(target=run, args=(i,semaphore))
    t.start()

运行后，可以看到5个一批的线程被放行。

3.3 事件Event

类名：Event

事件线程锁的运行机制：全局定义了一个Flag，如果Flag的值为False，那么当程序执行wait()方法时就会阻塞，如果Flag值为True，线程不再阻塞。这种锁，类似交通红绿灯（默认是红灯），它属于在红灯的时候一次性阻挡所有线程，在绿灯的时候，一次性放行所有排队中的线程。

事件主要提供了四个方法set()、wait()、clear()和is_set()。

调用clear()方法会将事件的Flag设置为False。

调用set()方法会将Flag设置为True。

调用wait()方法将等待“红绿灯”信号。

is_set():判断当前是否"绿灯放行"状态

下面是一个模拟红绿灯，然后汽车通行的例子：

#利用Event类模拟红绿灯
import threading
import time

event = threading.Event()

def lighter():
    green_time = 5       # 绿灯时间
    red_time = 5         # 红灯时间
    event.set()          # 初始设为绿灯
    while True:
        print("\33[32;0m 绿灯亮...\033[0m")
        time.sleep(green_time)
        event.clear()
        print("\33[31;0m 红灯亮...\033[0m")
        time.sleep(red_time)
        event.set()

def run(name):
    while True:
        if event.is_set():      # 判断当前是否"放行"状态
            print("一辆[%s] 呼啸开过..." % name)
            time.sleep(1)
        else:
            print("一辆[%s]开来，看到红灯，无奈的停下了..." % name)
            event.wait()
            print("[%s] 看到绿灯亮了，瞬间飞起....." % name)

if __name__ == '__main__':

    light = threading.Thread(target=lighter,)
    light.start()

    for name in ['奔驰', '宝马', '奥迪']:
        car = threading.Thread(target=run, args=(name,))
        car.start()

运行结果：

绿灯亮...
一辆[奔驰] 呼啸开过...
一辆[宝马] 呼啸开过...
一辆[奥迪] 呼啸开过...
一辆[奥迪] 呼啸开过...
......
 红灯亮...
一辆[宝马]开来，看到红灯，无奈的停下了...
一辆[奥迪]开来，看到红灯，无奈的停下了...
一辆[奔驰]开来，看到红灯，无奈的停下了...
绿灯亮...
[奥迪] 看到绿灯亮了，瞬间飞起.....
一辆[奥迪] 呼啸开过...
[奔驰] 看到绿灯亮了，瞬间飞起.....
一辆[奔驰] 呼啸开过...
[宝马] 看到绿灯亮了，瞬间飞起.....
一辆[宝马] 呼啸开过...
一辆[奥迪] 呼啸开过...
......