[ python ] 线程的操作

目录

（见右侧目录栏导航）

- 1. 前言
    - 1.1 进程
    - 1.2 有了进程为什么要有线程
    - 1.3 线程的出现
    - 1.4 进程和线程的关系
    - 1.5 线程的特点
- 2. 线程和python
    - 2.1 全局解释器锁（GIL）
    - 2.2 threading 模块
        - 2.2.1 进程的创建
        - 2.2.2 多线程和多进程的比较
        - 2.2.3 使用多线程实现socket聊天功能
        - 2.2.4 Thread类的其他方法
    - 2.3 守护线程
    - 2.4 锁
        - 2.4.1 死锁与递归锁
    - 2.5 信号量和事件
    - 2.6 条件和定时器
    - 2.7 线程队列
    - 2.8 线程池

1. 前言

1.1 进程

　　之前我们已经了解了操作系统中进程的概念，程序并不能单独运行，只有将程序装载到内存中，系统为它分配资源才能运行，而这种执行的程序就称之为进程。程序和进程的区别就在于：程序是指令的集合，它是进程运行的静态描述文本；进程是程序的一次执行活动，属于动态概念。在多道编程中，我们允许多个程序同时加载到内存中，在操作系统的调度下，可以实现并发地执行。这是这样的设计，大大提高了CPU的利用率。进程的出现让每个用户感觉到自己独享CPU，因此，进程就是为了在CPU上实现多道编程而提出的。

1.2 有了进程为什么要有线程

　　进程有很多优点，它提供了多道编程，让我们感觉我们每个人都拥有自己的CPU和其他资源，可以提高计算机的利用率。很多人就不理解了，既然进程这么优秀，为什么还要线程呢？其实，仔细观察就会发现进程还是有很多缺陷的，主要体现在两点上：

　　 　　(1) 进程只能在一个时间干一件事，如果想同时干两件事或多件事，进程就无能为力了。
       　　 (2) 进程在执行的过程中如果阻塞，例如等待输入，整个进程就会挂起，即使进程中有些工作不依赖于输入的数据，也将无法执行。

　　如果这两个缺点理解比较困难的话，举个现实的例子也许你就清楚了：如果把我们上课的过程看成一个进程的话，那么我们要做的是耳朵听老师讲课，手上还要记笔记，脑子还要思考问题，这样才能高效的完成听课的任务。而如果只提供进程这个机制的话，上面这三件事将不能同时执行，同一时间只能做一件事，听的时候就不能记笔记，也不能用脑子思考，这是其一；如果老师在黑板上写演算过程，我们开始记笔记，而老师突然有一步推不下去了，阻塞住了，他在那边思考着，而我们呢，也不能干其他事，即使你想趁此时思考一下刚才没听懂的一个问题都不行，这是其二。

　　现在你应该明白了进程的缺陷了，而解决的办法很简单，我们完全可以让听、写、思三个独立的过程，并行起来，这样很明显可以提高听课的效率。而实际的操作系统中，也同样引入了这种类似的机制——线程。

1.3 线程的出现

　　60年代，在OS中能拥有资源和独立运行的基本单位是进程，然而随着计算机技术的发展，进程出现了很多弊端，一是由于进程是资源拥有者，创建、撤消与切换存在较大的时空开销，因此需要引入轻型进程；二是由于对称多处理机（SMP）出现，可以满足多个运行单位，而多个进程并行开销过大。

    　　因此在80年代，出现了能独立运行的基本单位——线程（Threads）。
    
   　　 注意：进程是资源分配的最小单位，线程是CPU调度的最小单位；每个进程中至少有一个线程.

1.4 进程和线程的关系

　　进程和线程的区别可以归纳为以下 4点：

　　　　1. 地址空间和其他资源（如打开文件）：进程间相互独立，同一进程的各线程间共享。某进程内的线程在其他进程不可见；

　　　　2. 通信：进程间通信IPC，线程间可以直接读写进程数据段（如全局变量）来进行通信——需要进程同步和互斥锁的辅助，以保证数据的一致性；

　　　　3. 调度和切换：线程上下文切换比进程上下文切换要快的多；

　　　　4. 在多线程操作系统中，进程不是一个可执行的实体

　　通过漫画了解线程

1.5 线程的特点

　　在多线程的操作系统中，通常是在一个进程中包括多个线程，每个线程都是作为利用CPU的基本单位，是花费最小开销的实体。线程具有以下属性。

　　1. 轻型实体

　　　　线程中的实体基本上不拥有系统资源，只是有一点必不可少的、能保证独立运行的资源。

　　　　线程的实体包括程序、数据和TCB。线程是动态概念，它的动态特性由线程控制块TCB（Thread Control Block）描述。

    TCB包括以下信息：
　　（1）线程状态。
　　（2）当线程不运行时，被保存的现场资源。
　　（3）一组执行堆栈。
　　（4）存放每个线程的局部变量主存区。
　　（5）访问同一个进程中的主存和其它资源。

　　用于指示被执行指令序列的程序计数器、保留局部变量、少数状态参数和返回地址等的一组寄存器和堆栈。

　　2. 独立调度和分派的基本单位

　　　　在多线程OS中，线程是能独立运行的基本单位，因而也是独立调度和分派的基本单位。由于线程很“轻”，故线程的切换非常迅速且开销小（在同一进程中的）。

　　3. 共享进程资源

　　　　线程在同一进程中的各个线程，都可以共享该进程所拥有的资源，这首先表现在：所有线程都具有相同的进程id，这意味着，线程可以访问该进程的每一个内存资源；此外，还可以访问进程所拥有的已打开文件、定时器、信号量机构等。由于同一个进程内的线程共享内存和文件，所以线程之间互相通信不必调用内核。

　　4. 可并发性

　　　　在一个进程中的多个线程之间，可以并发执行，甚至允许在一个进程中所有线程都能并发执行；同样，不同进程中的线程也能并发执行，充分利用和发挥了处理机与外围设备并行工作的能力。

2. 线程和python

2.1 全局解释器锁（GIL）

　　Python代码的执行由Python虚拟机(也叫解释器主循环)来控制。Python在设计之初就考虑到要在主进程中，同时只有一个线程在执行。虽然 Python 解释器中可以“运行”多个线程，但在任意时刻只有一个线程在解释器中运行。对Python虚拟机的访问由全局解释器锁(GIL)来控制，正是这个锁能保证同一时刻只有一个线程在运行。

　　在多线程环境中，Python 虚拟机按以下方式执行：

　　　　1. 设置GIL；

　　　　2. 切换到一个线程运行；

　　　　3. 运行指定数量的字节码指令或线程主动让出控制；

　　　　4. 把线程设置为睡眠状态；

　　　　5. 解锁GIL；

　　　　6. 再次重复以上所有步骤。

　　所以说，同一时间，只有一个线程在CPU上运行。

2.2 threading 模块

　　threading 模块 和 multiprocessing 模块非常相似。

2.2.1 进程的创建

from threading import Thread
import time


def run_thread(name):
    time.sleep(0.5)
    print('hello,', name)


if __name__ == '__main__':
    t = Thread(target=run_thread, args=('xiaofei',))
    t.start()

from threading import Thread
import time


class Foo(Thread):
    def __init__(self, name):
        super(Foo, self).__init__()
        self.name = name

    def run(self):
        time.sleep(0.5)
        print('hello,', self.name)


if __name__ == '__main__':
    f = Foo('xiaofei')
    f.start()

2.2.2 多线程和多进程的比较

pid 对比

from multiprocessing import Process
from threading import Thread
import os

def func():
    print('child pid:', os.getpid())

if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    print('主进程pid:', os.getpid())
'''
主进程pid: 12068
child pid: 7764
'''
    t = Thread(target=func)
    t.start()
    print('主进程pid:', os.getpid())
'''
child pid: 820
主进程pid: 820
'''

通过对比发现父进程启动的子进程pid各自都有独立的pid，而对于线程来说，多个线程都是同一个主进程的pid，共享主进程的资源

开启效率的对比

from multiprocessing import Process
from threading import Thread


def func(name):
    print('hello,', name)


if __name__ == '__main__':
    p = Process(target=func, args=('xiaofei',))
    t = Thread(target=func, args=('xiaoA',))
    p.start()
    t.start()
    print('---主进程---')

对比执行结果：子线程比主进程执行要快，而子进程则比主进程要慢。

内存数据的共享问题

from multiprocessing import Process
from threading import Thread


def func():
    global n
    n = 0


if __name__ == '__main__':
    # --- 进程 ---
    n = 100
    p = Process(target=func)
    p.start()
    p.join()
    print(n)  # 已知进程间是相互独立的，子进程结束，子进程中的变量全部消失，这里还是主进程中定义的 n 为 100

    # --- 线程 ---
    n = 100
    t = Thread(target=func)  # 线程共享主进程中的变量
    t.start()
    t.join()
    print(n)  # 这里在子线程中被修改生效了。

2.2.3 使用多线程实现socket聊天功能

from threading import Thread
import socket


def func(conn):
    conn.send(b'welcome!')
    while True:
        res = conn.recv(1024)
        conn.send(res.upper())


if __name__ == '__main__':
    sk_server = socket.socket()
    sk_server.bind(('localhost', 8080))
    sk_server.listen(5)
    while True:
        conn, addr = sk_server.accept()
        t = Thread(target=func, args=(conn,))
        t.start()

import socket

sk_client = socket.socket()
sk_client.connect(('localhost', 8080))
res = sk_client.recv(1024)
print(res)
while True:
    inp = input('>>>').strip()
    if not inp: continue
    sk_client.send(inp.encode())
    print(sk_client.recv(1024).decode())

2.2.4 Thread类的其他方法

Thread实例对象的方法

isAlive(): 返回线程是否活动的；
getName(): 返回线程名；
setName(): 设置线程名。

threading模块提供的一些方法：

threading.currentThread(): 返回当前的线程变量；
threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程；
threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

join方法

from threading import Thread
import time


def func(name):
    time.sleep(0.5)
    print('hello,', name)


if __name__ == '__main__':
    t = Thread(target=func, args=('xiaofei',))
    t.start()
    t.join()  # join等待子线程执行完毕后在继续执行主进程中的代码
    print(t.is_alive())  # 子线程已经执行完毕，所有为False

'''
hello, xiaofei
False
'''

2.3 守护线程

无论是进程还是线程，都遵循：守护线/进程会等待主线/进程运行完毕后被销毁。需要强调的是：运行完毕并非终止运行。

　　1. 对于进程来说，运行完毕指的是主进程代码运行完毕；

　　2. 对于主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕。

详细解释：

　　1. 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收），然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源，才会结束；

　　2. 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

from threading import Thread
import time

def func():
    while True:
        time.sleep(3)
        print('hello world.')

if __name__ == '__main__':
    t = Thread(target=func)
    t.daemon = True
    t.start()

from threading import Thread
import time

def func():
    while True: # 死循环，但是它是守护线程，会随着主线程结束而终止
        print('hello world.')
        time.sleep(1)

def foo():
    for i in range(5):
        time.sleep(1)
        print('hello ', i)

if __name__ == '__main__':
    t1 = Thread(target=func)
    t2 = Thread(target=foo)
    t1.daemon = True    # 设置子线程 t1 为守护线程
    t1.start()
    t2.start()
    print('---主进程---')  # 主进程需要等待 t2 执行完毕才可以结束，而 t1 为守护进程，会随着主进程的结束而终止

2.4 锁

　　多线程和多进程最大的不同在于，多进程中，同一个变量，各自有一份拷贝存在于每个进程中，互不影响，而多线程中，所有变量都由所有线程共享，所以，任何一个变量都可以被任何一个线程修改，因此，线程之间共享数据最大的危险在于多个线程同时改一个变量，把内容给改乱了。

多个线程抢占资源的情况

from threading import Thread


def func(m):
    global n
    n = n - m
    n = n + m


def run_thread(m):
    for i in range(1000000):
        func(m)


if __name__ == '__main__':
    n = 100
    t1 = Thread(target=run_thread, args=(5,))
    t2 = Thread(target=run_thread, args=(8,))
    t1.start()
    t2.start()
    print(n)

上面的例子中，我们定义了一个共享变量 n ，初始值为 100 并且启动两个线程，先存后取，理论上结果应该为 100，但是由于线程的调度是由操作系统决定的，当t1, t2交替执行时，只有循环次数足够多，n 的结果就不一定是 0 了。

究其原因，是因为修改 n 需要多条语句，而执行这几条语句时，线程可能中断，从而导致多个线程把同一个对象的内容改乱了。

如果我们要确保 n 计算正确，就要给 func 上一把锁，当某个线程开始执行change_it()时，我们说，该线程因为获得了锁，因此其他线程不能同时执行change_it()，只能等待，直到锁被释放后，获得该锁以后才能改。由于锁只有一个，无论多少线程，同一时刻最多只有一个线程持有该锁，所以，不会造成冲突。创建一个锁就是

通过 threading.Lock() 来实现：

from threading import Thread, Lock

def func(m):
    global n
    n = n - m
    n = n + m

def run_thread(m, lock):
    for i in range(1000000):
        with lock:  # 使用关键字 with 自动获取、释放锁
            func(m)

if __name__ == '__main__':
    lock = Lock()
    n = 100
    t1 = Thread(target=run_thread, args=(5, lock))
    t2 = Thread(target=run_thread, args=(8, lock))
    t1.start()
    t2.start()
    for t in [t1, t2]: t.join()
    print(n)

因为python的线程虽然是真正的线程，但解释器执行代码时，有一个GIL锁（Global interpreter Lock）任何python线程执行前，必须先获取GIL锁，然后，每执行100条字节码，解释器就自动释放GIL锁，让别的线程有机会执行。这个GIL全局锁实际上把所有线程的执行代码给上了锁，所以，多线程在python中只能交替执行，即

使100个线程跑在100核CPU上，也只能用到1核。

GIL是Python解释器设计的历史遗留问题，通常我们用的解释器是官方实现的CPython，要真正利用多核，除非重写一个不带GIL的解释器。

所以，在Python中，可以使用多线程，但不要指望能有效利用多核。如果一定要通过多线程利用多核，那只能通过C扩展来实现，不过这样就失去了Python简单易用的特点。不过，也不用过于担心，Python虽然不能利用多线程实现多核任务，但可以通过多进程实现多核任务。多个Python进程有各自独立的GIL锁，互不影响。

2.4.1 死锁与递归锁

 所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁。

from threading import Thread, Lock

lock = Lock()
lock.acquire()
lock.acquire()
print('hello, world')
lock.release()
lock.release()

解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：

from threading import Thread, RLock

lock = RLock()
lock.acquire()
lock.acquire()
print('hello, world')
lock.release()
lock.release()

典型问题：科学家吃面

from threading import Thread, Lock
import time

noodle_lock = Lock()
fork_lock = Lock()


def eat1(name):
    noodle_lock.acquire()
    print('%s 抢到了面条。' % name)
    fork_lock.acquire()
    print('%s 抢到了叉子' % name)
    print('%s 吃面' % name)
    fork_lock.release()
    noodle_lock.release()


def eat2(name):
    fork_lock.acquire()
    print('%s 抢到了叉子' % name)
    time.sleep(0.5)
    noodle_lock.acquire()
    print('%s 抢到了面条' % name)
    print('%s 吃面' % name)
    noodle_lock.release()
    fork_lock.release()


for name in ['哪吒', 'egon', 'yuan']:
    t1 = Thread(target=eat1, args=(name,))
    t2 = Thread(target=eat2, args=(name,))
    t1.start()
    t2.start()

from threading import Thread, RLock
import time

fork_lock = noodle_lock = RLock()


def eat1(name):
    noodle_lock.acquire()
    print('%s 抢到了面条。' % name)
    fork_lock.acquire()
    print('%s 抢到了叉子' % name)
    print('%s 吃面' % name)
    fork_lock.release()
    noodle_lock.release()


def eat2(name):
    fork_lock.acquire()
    print('%s 抢到了叉子' % name)
    time.sleep(0.5)
    noodle_lock.acquire()
    print('%s 抢到了面条' % name)
    print('%s 吃面' % name)
    noodle_lock.release()
    fork_lock.release()


for name in ['哪吒', 'egon', 'yuan']:
    t1 = Thread(target=eat1, args=(name,))
    t2 = Thread(target=eat2, args=(name,))
    t1.start()
    t2.start()

2.5 信号量和事件

2.5.1 信号量

threading.Semaphore

线程和进程信号量和事件的使用方式是一样的。

from threading import Thread, Semaphore
import time

def ktv(sem, name):
    with sem:
        print('%s 走进ktv.' % name)
        time.sleep(3)
        print('%s 走出ktv.' % name)

if __name__ == '__main__':
    sem = Semaphore(3)
    for name in ['xiaofei', 'xiaoA', 'xiaoB', 'xiaoC', 'xiaoD']:
        t = Thread(target=ktv, args=(sem, name))
        t.start()

共享的摩拜单车就是一个很好信号量的例子，比如有 3 辆单车，现在有 5 个人要骑车，同一时间只能有3个人骑车，当3人中一人先完成，则第4个人就可以直接骑车，同一时间只能有3个人在骑车。

2.5.2 事件

threading.Event

from threading import Thread, Event
import time

def cars(e, name):
    if not e.is_set():
        print('33[31;1m[%s]车等待.33[0m' % name)
        e.wait()
    print('33[32;1m[%s]车通过.33[0m' % name)



def light(e):
    while True:
        if e.is_set():
            e.clear()
            print('33[31;1m红灯33[0m')
        else:
            e.set()
            print('33[32;1m绿灯33[0m')
        time.sleep(3)




if __name__ == '__main__':
    e = Event()
    t1 = Thread(target=light, args=(e, ))
    t1.start()
    for car in range(10):
        t2 = Thread(target=cars, args=(e, car))
        t2.start()
        time.sleep(1)