(十六) 学习笔记: Python线程相关

一、线程的相关概念

在一个进程内部要同时干很多件事,需要同时运行多个子任务 我们把进程内这些子任务称为线程，线程通常叫做轻量型的进程。
线程是最小的执行单元, 一个进程由至少一个线程组成。在单个程序中同时运行多个线程完成不同的工作，称为多线程，每个线程共享其所附属进程的所有资源。
但是线程不能够独立执行，必须依存在应用程序中，由应用程序提供多个线程执行控制。

进程与线程的区别
(1) 进程是资源分配的最小单位，线程是程序最小的执行单元,。
(2) 进程有独立的内存地址空间，每启动一个进程，系统就会为它分配内存地址，建立数据表来维护代码段、堆栈段和数据段。而线程是共享进程中的数据，使用相同的内存地址空间，因此CPU切换一个线程的花费远比进程要小的多，创建一个线程的开销也比进程要小很多。
(3) 线程之间更方便通信，同一进程下的线程共享所有资源。进程之间的通信需要以通信的方式（IPC)进行。
(4) 如果主线程结束，子线程也会结束，进程也就随之终止。而一个进程死掉，其他进程不受影响，所以多进程的程序要比多线程的程序健壮（稳定）。

多线程
一个进程可以由很多个线程组成，一般是由一个主线程和多个子线程组成，线程间共享进程的所有资源，每个线程也都有自己的堆栈和局部变量。通常线程由CPU独立调度执行，在多CPU环境下就允许多个线程同时运行。同样多线程也可以实现并发操作，每个请求分配一个线程来处理。

二、Python中的线程

在Python的标准库中提供了两个线程模块：_thread和threading，_thread模块，提供了低级别的、原始的线程，threading相对更高级一些，是对_thread进行了封装。绝大多数情况下，我们只需要使用threading这个高级模块。

(1) _thread模块

常用的方法：

_thread.start_new_thread(function, args[, kwargs])  开启子线程并返回它的ID。args为元组类型，kwargs为可选参数。
_thread.start_new(function,  args[, kwargs])  同上
_thread.interrupt_main()  抛出一个KeyboardInterrupt异常。子线程可以通过这个方法终止主线程执行。
_thread.exit()  结果当前线程。
_thread.exit_thread()  同上
_thread.allocate_lock()  返回一个锁对象。
_thread.get_ident()  返回当前进程的ID。

LockType锁对象的方法
lock.acquire(waitflag=1, timeout=-1)  获取锁。返回Boolean类型，True表示成功获取锁。
lock.release()  释放锁。
lock.locked()  返回锁的状态。True表示锁已被某个线程获取，False没有。

简单实例：

import _thread
import win32api
import time

def demo(name):
    print('子线程开始')
    print('子线程名：', name, "子线程标识符：", _thread.get_ident())
    print('子线程结束')

num = 0  # 锁用于控制对共享资源的访问

def lockdemo(lock):
    global num
    if lock.acquire():
        num += 1
        print(num)
        print('获取了锁，阻塞状态')
        time.sleep(2)
        print(lock.locked())  # True
        lock.release()  # 释放锁

if __name__ == '__main__':
    _threa.start_new_thread(demo, ('zhangsan', ))
    _thread.start_new_thread(demo, ('lisi', ))
    _thread.start_new_thread(demo, ('wangwu', ))

    time.sleep(3)  # 截停  让主线程听停止在这, 让子线程执行结束

    # 创建一个锁
    lock = _thread.allocate_lock()
    # 判断锁的状态
    print('初始状态：', lock.locked())  # False
    _thread.start_new_thread(lockdemo, (lock, ))
    _thread.start_new_thread(lockdemo, (lock, ))
    num += 1
    print('num:', num)  # 1  结果显示可以共享使用num变量，这一点与进程不同的

    time.sleep(10)
    print("结束了")

(2) threading模块 (推荐使用)

threading模块提供的类：
　　Local, Thread, Lock, Rlock, Condition, [Bounded]Semaphore, Event, Timer等。

对象	描述
Thread	表示一个执行线程的对象
Lock	锁原语对象
RLock	可重入锁对象，使单一线程可以（再次）获得已持有的锁（递归锁）
Condition	条件变量对象，使得一个线程等待另一个线程满足特定的“条件”，比如改变状态或
Event	条件变量的通用版本，任意数量的线程等待某个事件的发生，在该事件发生后所有
Semaphore	为线程间共享的有限资源提供了一个“计数器”，如果没有可用资源时会被阻塞
BoundedSemaphore	与Semaphore相似，不过它不允许超过初始值
Timer	与Thread相似，不过它要在运行前等待一段时间

threading模块下的方法

threading.active_count()  返回当前存在的Thread数量，包括主线程等。
threading.current_thread()  返回当前的Thread对象。
threading.get_ident()  返回当前线程的标识符ID。
threading.enumerate()  返回当前存在的Thread对象的列表，包括子线程，主线程及守护进程的线程。
threading.main_thread()  返回主线程对象。

实例：

import threading

if __name__ == '__main__':
    print(threading.active_count())  # 1
    print(threading.current_thread())  # <_MainThread(MainThread, started 7280)>
    print(threading.enumerate())  # [<_MainThread(MainThread, started 7280)>] 返回列表
    print(threading.main_thread())  # <_MainThread(MainThread, started 7280)>

(1) threading.Local对象

一个进程下的所有线程是共享内存空间的，有时候每个线程需要使用自己特定的数据。这时候只需创建一个Local(或一个子类)的实例并在其上存储属性即可。下面例子中你可以把local_val看成全局变量，但每个属性如local_val .name都是线程的局部变量，可以任意读写而互不干扰。
实例：

import threading

local_val = threading.local()  # 创建全局的local对象

def show():
    name = local_val.name  # 获取局部变量
    print("当前线程名:", threading.current_thread().name, "人名：", name)
   
def demo(name):
    local_val.name = name  # 设置局部变量
    show()

if __name__ == '__main__':
    p1 = threading.Thread(target= demo, args=('zhangsan',), name='进程A')
    p2 = threading.Thread(target= demo, args=('lisi',), name='进程B')
    p1.start()
    p2.start()
    p1.join()
    p2.join()

(2) threading.Thread对象

Thread对象的属性或方法

class threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)
参数：
  group 值为None, 线程组
  target 子线程执行的目标函数，会在run()方法中调用
  name 线程名称
  daemon 是否为守护进程（True/False）,默认为None从当前线程继承
方法：
  start() 启动进程，同时协调调用run()方法 ，每个线程对象最多只能调用一次
  run()  表示线程活动的方法，可以在子类中重写此方法
  is_alive()  返回线程是否处于活动状态。返回boolean
  join([timeout]) 等待线程终止。如果可选参数timeout是None（默认值），则该方法将阻塞，直到被调用join()方法的线程终止。如果timeout是一个正数，它最多会阻塞timeout秒。
　setName(name)/getName(): 设置/获取线程名。 

Thread对象创建的两种方法

法一：
import threading
import os 

def foo(arg):
      print("线程参数为:", arg, "线程name", threading.current_thread().name)

td1 = threading.Thread(target=foo, args=('zhangsan',), name='线程A')  # args传入的是元素必须加,号
td2 = threading.Thread(target=foo, args=('lisi',), name='线程B')
td3 = threading.Thread(target=foo, args=('wangwu',), name='线程C')

td1.start()
td2.start()
td3.start()
td1.join()
td2.join()
td3.join()
print('主线程结束了！')


法二（继承threading.Thread）：
需要重写__init__()和run()方法
import threading

class MyThread(threading.Thread):
    def __init__(self, arg):
        threading.Thread.__init__(self)
        self.arg = arg

    def run(self):
        print('调用子线程：', self.arg)

thread_list = []  # 存放子线程列表

for i in range(5):
    td = MyThread(i)
    td.start()
    thread_list.append(td)

for td in thread_list:
    td.join()  # 主线程阻塞要等待子线程结束。

print('主线程执行结束了')

执行结果：
调用子线程： 0
调用子线程： 1
调用子线程： 2
调用子线程： 3
调用子线程： 4
主线程执行结束了
[Finished in 0.8s]

注意：在法二中如果将join()写到第一个for循环里，将使多线程程序按照顺序执行。

(3) 互斥锁Lock对象

同步原语：一般在多线程代码中,总会有一些特定的函数或代码块不希望(或不应该)被多个线程同时执行,通常包括修改数据库、更新文件或其他会产生竞态条件的类似情况。如果两个线程运行的顺序发生变化,就有可能造成代码的执行轨迹或行为不相同,或者产生不一致的数据。这就是需要使用同步的情况。当任意数量的线程可以访问临界区的代码,但在给定的时刻只有一个线程可以通过时,就是使用同步的时候了。我们可以选择适合的同步原语,或者线程控制机制来执行同步。锁是所有机制中最简单、最低级的机制,而信号量用于多线程竞争有限资源的情况。

临界资源：一次仅允许一个进程使用的资源称为临界资源。许多物理设备都属于临界资源，如输入机、打印机等。

临界区：临界区指的是一个访问临界资源的程序片段。当有线程进入临界区段时，其他线程必须等待，有一些同步的机制必须在临界区段的进入点与离开点实现，以确保这些共用资源是被互斥获得使用。

作用：为了多个线程同时操作一个内存中的资源时不产生混乱，我们使用锁。
锁有两种状态： locked和unlocked
使用Lock.acruire()方法 将锁设置为locked的状态
使用Lock.release()方法 将锁设置为unlocked的状态

注意：

1. 如果多线程直接共用这个锁，可能会出现死锁的情况。
2. 当前线程锁定后其他线程会等待(线程等待/线程阻塞)
3. 不能进行重复锁定

常用方法

import threading
lock = threading.Lock()
Lock.acquire(blocking=True, timeout=-1)  # 对资源加锁,锁定成功返回True 
Lock.release()  # 解锁

简写使用with的形式

 if lockA.acquire():  # 成功获得锁继续执行下面语句，没有锁住成功就一直阻塞等待
        for i in range(101):  # 单线程执行
            num+=1 
        mutex.release() #释放锁 

同上，使用with，自动获取和释放锁

with lockA:
    for i in range(101):   # 单线程执行
          num+=1 

未加锁时，可能出现线程冲突的问题:

我们开启5个子线程，让每一个线程从0加到100
import threading

num = 0

class MyThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        global num
        for i in range(101):
            num += i
        print(num)
        num = 0  # 让累加值重置为0

for i in range(5):
    my = MyThread()  # 开启5个子线程
    my.start()

我们期望的是输出5次0-100的加和5050，但结果是这样

使用Lock加锁后

num = 0
lock = threading.Lock()

class MyThread(threading.Thread): 
    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        global num
        if lock.acquire():  # 开启锁 当锁定成功之后其他线程会阻塞
            for i in range(101):
                num += i
            print(num)  
            lock.release()  # 释放锁
        num = 0

myList = []
for i in range(5):  # 开启五个子线程
    my = MyThread()
    my.start()
    myList.append(my)

for my in myList:
    my.join()  # 线程等待
print("主线程执行结束了")

结果为

线程死锁

(1) 实例1

num = 0
lock = threading.Lock()

class MyThread(threading.Thread):
  def run(self):
    global num
    time.sleep(1)
    if lock.acquire():
      num = num + 1
      print(num)
      lock.acquire()  # 造成死锁
      lock.release()
      lock.release()

for i in range(5):
    t = MyThread()
    t.start()
分析：线程第一次请求资源，请求后还未 release ，再次acquire，最终无法释放，造成死锁。

(2) 实例2

class MyThread(threading.Thread):
  def func1(self):
    if lockA.acquire():
        print(self.name, "got lockA")
        if lockB.acquire():
            print(self.name, "got lockB")
            lockB.release()
        lockA.release()

  def func2(self):
    if lockB.acquire():
        print(self.name, "got lockB")
        if lockA.acquire():
            print(self.name, "got lockA")
            lockA.release()
        lockB.release()

  def run(self):
    self.func1()
    self.func2()

lockA = threading.Lock()
lockB = threading.Lock()

for i in range(5):
    t = MyThread()
    t.start()
分析：互相调用产生的死锁，是两个函数中都会调用相同的资源，互相等待对方结束。如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁。

(4) 可重入锁RLock

RLock允许在同一线程中被多次acquire()。如果使用RLock，那么acquire()和release()必须成对出现， 调用了几次acquire()锁请求，则还必须调用几次的release()才能在线程中释放锁对象。
一个线程所有的acquire()都被release()，其他的线程才能获得资源。
实例：

上述死锁例子1中如果换为可重入锁，则不会发生死锁的问题
num = 0
lock = threading.RLock()

class MyThread(threading.Thread):
  def run(self):
    global num
    time.sleep(1)
    if lock.acquire():
      num = num + 1
      print(num)
      lock.acquire()  # ok 
      lock.release()
      lock.release()

for i in range(5):
    t = MyThread()
    t.start()

简写使用With的形式

rlock = threading.RLock()

class MyThread(threading.Thread):
    def run(self):
        with rlock:    # 简写
            print("第一次锁定")
            with rlock:    # 简写
                print("第二次锁定")
                print(self.name)  # 打印thread name
                time.sleep(2)
        print("我是锁外面正常执行的代码")  # 其他线程阻塞不输出这行

for i in range(5):  # 开启5个子线程
    MyThread().start()

(5) 信号量Semaphore

参考资料：
https://docs.python.org/3.6/library/_thread.html
http://python.jobbole.com/82723/
https://www.cnblogs.com/tkqasn/p/5700281.html
https://docs.python.org/3.6/library/threading.html
https://www.liaoxuefeng.com/wiki/0014316089557264a6b348958f449949df42a6d3a2e542c000