并发编程-线程

1. 线程
2. GIL
3. 守护线程
4. 线程锁（互斥锁 and 递归锁）
5. 信号量
6. 事件
7. 条件
8. 定时器

1.线程：

• 特点
  在多线程的操作系统中，通常是在一个进程中包括多个线程，每个线程都是作为利用CPU的基本单位，是花费最小开销的实体。线程具有以下属性。
  　　1）轻型实体
  　　线程中的实体基本上不拥有系统资源，只是有一点必不可少的、能保证独立运行的资源。
  　　线程的实体包括程序、数据和TCB。线程是动态概念，它的动态特性由线程控制块TCB（Thread Control Block）描述。
  　　2）独立调度和分派的基本单位。
  　　在多线程OS中，线程是能独立运行的基本单位，因而也是独立调度和分派的基本单位。由于线程很“轻”，故线程的切换非常迅速且开销小（在同一进程中的）。
  　　3）共享进程资源。
  　　线程在同一进程中的各个线程，都可以共享该进程所拥有的资源，这首先表现在：所有线程都具有相同的进程id，这意味着，线程可以访问该进程的每一个内存资源；此外，还可以访问进程所拥有的已打开文件、定时器、信号量机构等。由于同一个进程内的线程共享内存和文件，所以线程之间互相通信不必调用内核。
  　 4）可并发执行
  在一个进程中的多个线程之间，可以并发执行，甚至允许在一个进程中所有线程都能并发执行；同样，不同进程中的线程也能并发执行，充分利用和发挥了处理机与外围设备并行工作的能力。

1. 线程是进程中的执行单位
2. 线程是cpu执行的最小单位
3. 线程之间资源共享
4. 线程的开启和关闭、切换的时空开销远远小于进程

2.GIL 全局解释锁

Cpython解释器在解释代码过程中容易产生数据不安全问题（Cpython的特性，在Jpython就没有这个问题）
GIL，锁的是线程，同一时刻只有一个线程能够访问cpu

• 如果刚好遇上cpu时间片轮转或者线程阻塞，线程会挂起，然后归还锁，让其他线程访问cpu进行计算。则会造成数据安全问题，此时需要线程锁来保证数据安全。

import time
from threading import Thread, Lock

def func(lock):
    lock.acquire()
    global n
    temp = n
    time.sleep(0.2)  # 线程阻塞
    n = temp - 1
    lock.release()


n = 10
t_list = []
lock = Lock()
for i in range(10):
    t = Thread(target=func, args=(lock, ))
    t.start()
    t_list.append(t)
for t in t_list:t.join() # 等待所有线程执行完毕
print(n)

3.守护线程

1. 主线程会等待所有非守护线程结束后才算运行完毕。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收。
2. 在主线程结束(其他非主线程非守护线程全部结束)后，守护线程会立即结束。

• 与守护进程的区别
  守护进程在主进程代码执行完毕后，就会立即结束其生命周期，不管其他子进程是否运行完毕。（守护进程在主进程代码执行完后立即结束，主进程与最慢结束的进程一起结束）
  守护线程会等待主线程结束的时候，才会结束。而主线程的结束需要等待其他子线程全部结束，整体资源都被回收后才结束。（守护线程与主线程要等待其他线程结束后才最后结束）

from threading import Thread
import time

def func():
    print('子线程 is running')
    time.sleep(5)
def func2():
    while True:
        print('守护线程 is running')
        time.sleep(1)

# 开启线程
t1 = Thread(target=func)
t1.start()
t2 = Thread(target=func2)
t2.daemon = True  # t2转为守护线程
t2.start()
print('主线程结束')

4.线程锁

互斥锁

一个线程中有多个锁的时候，可能会形成死锁问题。需要用到递归锁来解决

# 互斥锁 Lock
# 科学家吃面问题  需要 拿到 叉 和 面 才能吃到面
from threading import Lock, Thread
import time
fork_lock = Lock()
noodle_lock = Lock()
def eat1(name):
    fork_lock.acquire()
    print('%s拿到叉子了'%name)
    noodle_lock.acquire()
    print('%s拿到面条了'%name)
    print('%s开始吃面了'%name)
    noodle_lock.release()
    fork_lock.release()

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

Thread(target=eat1, args=('jobs', )).start()
Thread(target=eat2, args=('steve', )).start()
Thread(target=eat1, args=('bob', )).start()
# 1. jobs先拿到叉子和面条，开始吃面，然后归还了锁
# 2. steve拿到面条，阻塞一秒，阻塞的同时 bob拿到了叉子， 由于需要同时拿到两把锁才能开始吃面，所以此时程序开始挂起进入死锁。

递归锁

1. 为了解决死锁问题
2. 在一个线程里可以被acquire多次

from threading import RLock, Thread
import time
fork_lock = noodle_lock = RLock()
def eat1(name):
    fork_lock.acquire()
    print('%s拿到叉子了'%name)
    noodle_lock.acquire()
    print('%s拿到面条了'%name)
    print('%s开始吃面了'%name)
    noodle_lock.release()
    fork_lock.release()

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

Thread(target=eat1, args=('jobs', )).start()
Thread(target=eat2, args=('steve', )).start()
Thread(target=eat1, args=('bob', )).start()

5.信号量

与进程一样，一次只能有指定的数量的线程来操作一套资源。

from threading import Semaphore, Thread
import time
def func(a, b, sem):
    sem.acquire()
    time.sleep(1)
    print(a+b)
    sem.release()

sem = Semaphore(4)
for i in range(10):
    Thread(target=func, args=(i, i+5, sem)).start()

6.事件

e.is_set()：查看当前是否为阻塞状态，False则为阻塞
e.wait()：根据事件的状态决定自己是否在wait处阻塞。如果当前为阻塞，则一直等待
e.clear()：将当前事件变成阻塞
e.set():将当前事件解除阻塞

# 模拟 连接数据库
import time
import random
from threading import Thread, Event

def connect_db(e):
    try:
        count = 0  # 设置连接次数
        while count < 3:
            e.wait(0.5)  # 状态为false，阻塞状态只等待 timeout 参数的时间
            if e.is_set() == True:
                print('连接数据库成功')
                break
            else:
                count += 1
                print('第%s次连接失败'%count)
        else:
            raise TimeoutError('数据库连接失败')
    except TimeoutError as e:
        print(e)



def check_web(e):
    time.sleep(random.randint(0, 3))
    e.set()  # 将事件解除阻塞


e =Event()
t1 = Thread(target=check_web, args=(e, ))
t2 = Thread(target=connect_db, args=(e, ))
t1.start()
t2.start()

7.条件

一个条件创建之初，默认有一个False状态，wait在False状态会一直等待，等到notify

• acquire
  通过acquire拿到锁，然后wait一直在等待钥匙
• release
  释放锁
• wait()
  在条件创建之初，默认有一个False状态，wait在False状态会一直等待，条件的状态改变，则继续执行之后的代码
• notify(int数据类型)
  相当于之前拿到的锁，在等待notify制造的钥匙。notify的参数就是钥匙的个数，条件创建时得False状态变为True。
  根据notify的参数，执行相应个数的线程

from threading import Condition, Thread

def func(con, i):
    con.acquire()
    con.wait()
    print('当前在第%s个循环中'%i)
    con.release()


con = Condition()
for i in range(10):
    Thread(target=func, args=(con, i)).start()
while True:
    num = int(input('>>>'))
    con.acquire()
    con.notify(num)
    con.release()

9.定时器

定时开启一个线程
Timer(time, func).start()

# 每隔5秒，开启一个线程执行func中的代码。
from threading import Timer
import time


def func():
    print('时间同步相关code')


while True:
    Timer(5, func).start()
    time.sleep(5)
# 由于while True会一直快速执行以下的代码，需要在while True下面添加sleep(5)
# 这样就是完整的5秒，不多不少。 如果sleep(5)在func中添加，则会加上程序的运行时间，时间会超过5秒

threading模块