python3之线程threading_python3 threading

本章节分享进程threading，分享给刚学python的小伙伴，一起学习，共同进步

什么是线程：

线程（Thread）也叫轻量级进程，是操作系统能够进行运算调度的最小单位，它被包涵在进程之中，是进程中的实际运作单位。线程自己不拥有系统资源，只拥有一点儿在运行中必不可少的资源，但它可与同属一个进程的其它线程共享进程所拥有的全部资源。一个线程可以创建和撤消另一个线程，同一进程中的多个线程之间可以并发执行。
举个简单的例子来理解下：
假定有一 7 * 24 小时不停工的工厂，由于其电力有限，一次仅供一个车间使用，当一个车间在生产时，其他车间停工。在这里我们可以理解这个工厂相当于操作系统，供电设备相当于 CPU，一个车间相当于一个进程。

一个车间里，可以有很多工人。他们协同完成一个任务。车间的空间是工人们共享的，这里一个工人就相当于一个线程，一个进程可以包括多个线程。比如许多房间是每个工人都可以进出的。这象征一个进程的内存空间是共享的，每个线程都可以使用这些共享内存。

有时候资源有限，比如有些房间最多只能容纳一个人，当一个人占用的时候，其他人就不能进去，只能等待。这代表一个线程使用某些共享内存时，其他线程必须等它结束，才能使用这一块内存。

一个防止他人进入的简单方法，就是门口加一把锁。先到的人锁上门，后到的人看到上锁，就在门口排队，等锁打开再进去。这就叫”互斥锁”（Mutual exclusion，缩写 Mutex ），防止多个线程同时读写某一块内存区域。
还有些房间，可以同时容纳 n 个人，比如厨房。也就是说，如果人数大于 n，多出来的人只能在外面等着。这好比某些内存区域，只能供给固定数目的线程使用。这时的解决方法，就是在门口挂 n 把钥匙。进去的人就取一把钥匙，出来时再把钥匙挂回原处。后到的人发现钥匙架空了，就知道必须在门口排队等着了。这种做法叫做”信号量”（ Semaphore ），用来保证多个线程不会互相冲突。

不难看出， mutex 是 semaphore 的一种特殊情况（n=1时）。也就是说，完全可以用后者替代前者。但是，因为 mutex 较为简单，且效率高，所以在必须保证资源独占的情况下，还是采用这种设计。

线程有 就绪、阻塞、运行 三种基本状态。

1. 就绪状态是指线程具备运行的所有条件，逻辑上可以运行，在等待处理机；
2. 运行状态是指线程占有处理机正在运行；
3. 阻塞状态是指线程在等待一个事件（如某个信号量），逻辑上不可执行。

三种状态的相互转化如下图所示：

threading函数

在Python3中方法名和函数名统一成了以字母小写加下划线的命令方式，但是Python2.x中threading模块的某些以驼峰命名的方法和函数仍然可用，如threading.active_count()和threading.activeCount()是一样的。

通常情况下，Python程序启动时，Python解释器会启动一个继承自threading.Thread的threading._MainThread线程对象作为主线程，所以涉及到threading.Thread的方法和函数时通常都算上了这个主线程的，比如在启动程序时打印threading.active_count()的结果就已经是1了。

threading.active_count()：返回当前存活的threading.Thread线程对象数量，等同于len(threading.enumerate())。
threading.current_thread()：返回此函数的调用者控制的threading.Thread线程对象。如果当前调用者控制的线程不是通过threading.Thread创建的，则返回一个功能受限的虚拟线程对象。
threading.get_ident()：返回当前线程的线程标识符。注意当一个线程退出时，它的线程标识符可能会被之后新创建的线程复用。
threading.enumerate()：返回当前存活的threading.Thread线程对象列表。
threading.main_thread()：返回主线程对象，通常情况下，就是程序启动时Python解释器创建的threading._MainThread线程对象。
threading.stack_size([size])：返回创建线程时使用的堆栈大小。也可以使用可选参数size指定之后创建线程时的堆栈大小，size可以是0或者一个不小于32KiB的正整数。如果参数没有指定，则默认为0。如果系统或者其他原因不支持改变堆栈大小，则会报RuntimeError错误；如果指定的堆栈大小不合法，则会报ValueError，但并不会修改这个堆栈的大小。32KiB是保证能解释器运行的最小堆栈大小，当然这个值会因为系统或者其他原因有限制，比如它要求的值是大于32KiB的某个值，只需根据要求修改即可。
1
2
3
4
5
6

线程对象：threading.Thread

threading.Thread目前还没有优先级和线程组的功能，而且创建的线程也不能被销毁、停止、暂定、恢复或中断。

守护线程：只有所有守护线程都结束，整个Python程序才会退出，但并不是说Python程序会等待守护线程运行完毕，相反，当程序退出时，如果还有守护线程在运行，程序会去强制终结所有守护线程，当守所有护线程都终结后，程序才会真正退出。可以通过修改daemon属性或者初始化线程时指定daemon参数来指定某个线程为守护线程。

非守护线程：一般创建的线程默认就是非守护线程，包括主线程也是，即在Python程序退出时，如果还有非守护线程在运行，程序会等待直到所有非守护线程都结束后才会退出。

注：守护线程会在程序关闭时突然关闭（如果守护线程在程序关闭时还在运行），它们占用的资源可能没有被正确释放，比如正在修改文档内容等，需要谨慎使用。

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

如果这个类的初始化方法被重写，请确保在重写的初始化方法中做任何事之前先调用threading.Thread类的__init__方法。

    group：应该设为None，即不用设置，使用默认值就好，因为这个参数是为了以后实现ThreadGroup类而保留的。
    target：在run方法中调用的可调用对象，即需要开启线程的可调用对象，比如函数或方法。
    name：线程名称，默认为“Thread-N”形式的名称，N为较小的十进制数。
    args：在参数target中传入的可调用对象的参数元组，默认为空元组()。
    kwargs：在参数target中传入的可调用对象的关键字参数字典，默认为空字典{}。
    setDaemon：默认为None，即继承当前调用者线程（即开启线程的线程，一般就是主线程）的守护模式属性，如果不为None，则无论该线程是否为守护模式，都会被设置为“守护模式”。
start()：开启线程活动。它将使得run()方法在一个独立的控制线程中被调用，需要注意的是同一个线程对象的start()方法只能被调用一次，如果调用多次，则会报RuntimeError错误。
run()：此方法代表线程活动。
join(timeout=None)：让当前调用者线程（即开启线程的线程，一般就是主线程）等待，直到线程结束（无论它是什么原因结束的），timeout参数是以秒为单位的浮点数，用于设置操作超时的时间，返回值为None。如果想要判断线程是否超时，只能通过线程的is_alive方法来进行判断。join方法可以被调用多次。如果对当前线程使用join方法（即线程在内部调用自己的join方法），或者在线程没有开始前使用join方法，都会报RuntimeError错误。
name：线程的名称字符串，并没有什么实际含义，多个线程可以赋予相同的名称，初始值由初始化方法来设置。
ident：线程的标识符，如果线程还没有启动，则为None。ident是一个非零整数，参见threading.get_ident()函数。当线程结束后，它的ident可能被其他新创建的线程复用，当然就算该线程结束了，它的ident依旧是可用的。
is_alive()：线程是否存活，返回True或者False。在线程的run()运行之后直到run()结束，该方法返回True。
setDaemon：表示该线程是否是守护线程，True或者False。设置一个线程的setdDaemon必须在**线程的start()方法**之前，否则会报RuntimeError错误。这个值默认继承自创建它的线程，主线程默认是非守护线程的，所以在主线程中创建的线程默认都是非守护线程的，即setDaemon=False。
1
2
3
4
5
6
7
8
9
10
11
12
13

import threading
import time
def run(n):
    print("task:", n, threading.current_thread())
    time.sleep(2)
   	print("task done", n)
   	
start_time = time.time()

t_obj = []
for i in range(50):
    t = threading.Thread(target=run,args=("t-%d"%i,))
    # 把当前线程设置为守护线程,当主线程执行完毕则关闭所有的子线程
    # t.setDaemon(True)
    t.start()
    # 为了防止不阻塞后面的子进程启动,不在这里加join,先放到一个列表中
    t_obj.append(t)

# 可以保证所有的子进程都进行完毕了再进入主进程
for t in t_obj:
    t.join()

# 主线程
# threading.activeCount() 当前活跃的进程数
print("----all thread has finished----",threading.current_thread(),threading.activeCount())
print("cost:",time.time() - start_time)
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

锁对象：threading.Lock

threading.Lock是直接通过_thread模块扩展实现的。

当锁在被锁定时，它并不属于某一个特定的线程。

锁只有“锁定”和“非锁定”两种状态，当锁被创建时，是处于“非锁定”状态的。当锁已经被锁定时，其他线程再次调用acquire()方法会被阻塞执行，直到锁被获得锁的线程调用release()方法释放掉锁并将其状态改为“非锁定”。

同一个线程获取锁后，如果在释放锁之前再次获取锁会导致当前线程阻塞，除非有另外的线程来释放锁，如果只有一个线程，并且发生了这种情况，会导致这个线程一直阻塞下去，即形成了死锁。所以在获取锁时需要保证锁已经被释放掉了，或者使用递归锁来解决这种情况。

acquire(blocking=True, timeout=-1)：获取锁，并将锁的状态改为“锁定”，成功返回True，失败返回False。当一个线程获得锁时，会阻塞其他尝试获取锁的线程，直到这个锁被释放掉。timeout默认值为-1，即将无限阻塞等待直到获得锁，如果设为其他的值时（单位为秒的浮点数），将最多阻塞等待timeout指定的秒数。当blocking为False时，timeout参数被忽略，即没有获得锁也不进行阻塞。
release()：释放一个锁，并将其状态改为“非锁定”，需要注意的是任何线程都可以释放锁，不只是获得锁的线程（因为锁不属于特定的线程）。release()方法只能在锁处于“锁定”状态时调用，如果在“非锁定”状态时调用则会报RuntimeError错误。
1
2

import threading
import time
def run(n):
    lock.acquire()
    print("task:",n,threading.current_thread())
    lock.release()

lock = threading.Lock()

start_time = time.time()

t_obj = []
for i in range(50):
    t = threading.Thread(target=run,args=("t-%d"%i,))
    # 把当前线程设置为守护线程,当主线程执行完毕则关闭所有的子线程
    # t.setDaemon(True)
    t.start()
    # 为了防止不阻塞后面的子进程启动,不在这里加join,先放到一个列表中
    t_obj.append(t)

# 可以保证所有的子进程都进行完毕了再进入主进程
for t in t_obj:
    t.join()

# 主线程
# threading.activeCount() 当前活跃的进程数
print("----all thread has finished----",threading.current_thread(),threading.activeCount())
print("cost:",time.time() - start_time)
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

递归锁对象：threading.RLock

递归锁和普通锁的差别在于加入了“所属线程”和“递归等级”的概念，释放锁必须有获取锁的线程来进行释放，同时，同一个线程在释放锁之前再次获取锁将不会阻塞当前线程，只是在锁的递归等级上加了1（获得锁时的初始递归等级为1）。

使用普通锁时，对于一些可能造成死锁的情况，可以考虑使用递归锁来解决。

acquire(blocking=True, timeout=-1)：与普通锁的不同之处在于：当使用默认值时，如果这个线程已经拥有锁，那么锁的递归等级加1。线程获得锁时，该锁的递归等级被初始化为1。当多个线程被阻塞时，只有一个线程能在锁被解时获得锁，这种情况下，acquire()是没有返回值的。
release()：没有返回值，调用一次则递归等级减1，递归等级为零时表示这个线程的锁已经被释放掉，其他线程可以获取锁了。可能在一个线程中调用了多次acquire()，导致锁的递归等级大于了1，那么就需要调用对应次数的release()来完全释放锁，并将它的递归等级减到零，其他的线程才能获取锁，不然就会一直被阻塞着。
1
2

import threading, time
def run1():
    print("grab the first part data")
    lock.acquire()
    global num
    num += 1
    lock.release()
    return num
def run2():
    print("grab the second part data")
    lock.acquire()
    global num2
    num2 += 1
    lock.release()
    return num2
def run3():
    lock.acquire()
    res = run1()
    print('--------between run1 and run2-----')
    res2 = run2()
    lock.release()
    print(res, res2)

num, num2 = 0, 0
# 递归锁防止锁死
lock = threading.RLock()
for i in range(10):
    t = threading.Thread(target=run3)
    t.start()

while threading.active_count() != 1:
    print(threading.active_count())
else:
    print('----all threads done---')
    print(num, num2)
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

信号量

信号量用来控制线程并发数的，BoundedSemaphore或Semaphore管理一个内置的计数器，每当调用acquire()时-1，调用release()时+1。

计数器不能小于0，当计数器为0时，acquire()将阻塞线程至同步锁定状态，直到其他线程调用release()。(类似于停车位的概念)

  BoundedSemaphore与Semaphore的唯一区别在于前者将在调用release()时检查计数器的值是否超过了计数器的初始值，如果超过了将抛出一个异常。
1

import threading,time

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

semaphore = threading.BoundedSemaphore(5)

if __name__ == "__main__":
    semaphore = threading.BoundedSemaphore(5)  # 最多允许5个线程同时运行
    for i in range(22):
        t = threading.Thread(target=run, args=(i,))
        t.start()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

事件event

同进程的一样,线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行

Event几种方法:

event.isSet()：返回event的状态值；

event.wait()：如果 event.isSet()==False将阻塞线程；

event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；

event.clear()：恢复event的状态值为False。

import threading,time
event = threading.Event()
def lighter():
    count = 0
    # 先设置路灯
    event.set()
    while True:
        # 红灯
        if count >= 5 and count < 10:
            event.clear()
            print("\033[41;1mred light is on...\033[0m")
        elif count > 10:
            event.set()
            count = 0
        else:
            print("\033[42;1mgreen light is on...\033[0m")
        time.sleep(1)
        count += 1

def car(name):
    while True:
        if event.is_set():
            print("[%s] running..." % name)
            time.sleep(1)
        else:
            print("[%s] sees red light , waiting...." % name)
            event.wait()
            print("\033[34;1m[%s] green light is on, start going...\033[0m" % name)

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

car = threading.Thread(target=car,args=("长安",))
car.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

队列（Queue）

Python的Queue模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出)队列Queue，LIFO（后入先出）队列LifoQueue，和优先级队列PriorityQueue。这些队列都实现了锁原语，能够在多线程中直接使用。可以使用队列来实现线程间的同步。

常用方法：

Queue.qsize() 返回队列的大小
Queue.empty() 如果队列为空，返回True,反之False
Queue.full() 如果队列满了，返回True,反之False，Queue.full 与 maxsize 大小对应
Queue.get([block[, timeout]])获取队列，timeout等待时间
Queue.get_nowait() 相当于Queue.get(False)，非阻塞方法
Queue.put(item) 写入队列，timeout等待时间
Queue.task_done() 在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号。每个get()调用得到一个任务，接下来task_done()调用告诉队列该任务已经处理完毕。
Queue.join() 实际上意味着等到队列为空，再执行别的操作
1
2
3
4
5
6
7
8

import queue,time,threading

q = queue.Queue(maxsize=10)

def Product(name):
    count = 1
    while True:
        q.put("骨头%s" % count)
        print("生产了骨头", count)
        count += 1
        time.sleep(0.1)

def Consumer(name):
    # while q.qsize() > 0:
    while True:
        print("[%s] 取到[%s] 并且吃了它..." % (name, q.get()))
        time.sleep(0.5)

p1 = threading.Thread(target=Product,args=("bob",))
c1 = threading.Thread(target=Consumer,args=("andy",))
c2 = threading.Thread(target=Consumer,args=("james",))

p1.start()
c1.start()
c2.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

Python 中的多线程之 GIL

说到 Python 中的多线程，一个绕不过去的话题就是全局锁 GIL(Global interpreter lock)。GIL 限制了同一时刻只能有一个线程运行，无法发挥多核 CPU 的优势。首先需要明确的一点是 GIL 并不是 Python 的特性，它是在实现 Python 解析器(CPython)时所引入的一个概念。就好比 C++ 是一套语言（语法）标准，但是可以用不同的编译器来编译成可执行代码。有名的编译器例如GCC，INTEL C++ ，Visual C++等。Python 也一样，同样一段代码可以通过 CPython，PyPy，Psyco 等不同的 Python 执行环境来执行。像其中的 JPython 就没有GIL。然而因为 CPython 是大部分环境下默认的 Python 执行环境。所以在很多人的概念里 CPython 就是 Python，也就想当然的把 GIL 归结为 Python 语言的缺陷。所以这里要先明确一点：GIL 并不是 Python 的特性，Python 完全可以不依赖于 GIL。

GIL 本质就是一把互斥锁，既然是互斥锁，所有互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。在一个 Python 的进程内，不仅有主线程或者由该主线程开启的其他线程，还有解释器开启的垃圾回收等解释器级别的线程，总之，所有线程都运行在这一个进程内，所有数据都是共享的，这其中，代码作为一种数据也是被所有线程共享的，多个线程先访问到解释器的代码，即拿到执行权限，然后将 target 的代码交给解释器的代码去执行,
解释器的代码是所有线程共享的，所以垃圾回收线程也可能访问到解释器的代码而去执行，这就导致了一个问题:对于同一个数据 100，可能线程 1 执行 x=100 的同时，而垃圾回收执行的是回收 100 的操作，解决这种问题没有什么高明的方法，就是加锁处理，即 GIL。
因此，有了 GIL 的存在，同一时刻同一进程中只有一个线程被执行，那么有人可能要问了：进程可以利用多核，而 Python 的多线程 却无法利用多核优势，Python 的多线程是不是没用了？
答案当然不是。
首先明确我们线程执行的任务是什么，是做计算（计算密集型）还是做输入输出（I/O 密集型），不同地场景使用不同的方法。 多核 CPU，意味着可以有多个核并行完成计算，所以多核提升的是计算性能,但每个 CPU 一旦遇到 I/O 阻塞，仍然需要等待，所以多核对 I/O 密集型任务没什么太高提升。

在 Python 中，对于计算密集型任务，多进程占优势，对于 I/O 密集型任务（I/O不占cpu），多线程占优势。