大家好，又见面了，我是你们的朋友全栈君。

线程同步与线程锁

线程同步

概念
* 线程同步，线程间协同，通过某种技术，让一个线程访问某些数据时，其他线程不能访问这些数据，直到该线程完 成对数据的操作。

1.threading.Event对象

• Event事件，是线程间通信机制中最简单的实现，使用一个内部的标记flag,通过flag的True或False的变化 来进行操作

• 老板雇佣了一个工人，让他生产杯子，老板一直等着这个工人，直到生产了10个杯子

import threading
import time
import logging

logging.basicConfig(format="%(asctime)s %(threadName)s %(thread)s %(message)s",level=logging.INFO)

def worker(event:threading.Event,count = 10):
    logging.info("我是worker工作线程")
    cups = []
    while True:
        logging.info("制作了一个 cup")
        time.sleep(0.2)
        cups.append(1)
        if len(cups)>=count:
            event.set()
            break
    logging.info("制作完成：{}".format(cups))

def boss(event:threading.Event):
    logging.info("我是boss")
    event.wait()
    logging.info("Good Job")

event = threading.Event()
b = threading.Thread(target=boss,args=(event,))
w = threading.Thread(target=worker,args=(event,))
b.start()
w.start()

• 使用同一个Event对象的标记flag。

• 谁wait就是等到flag变为True，或等到超时返回False。不限制等待的个数。

• wait的使用

from threading import Thread,Event
import logging

logging.basicConfig(format="%(asctime)s %(threadName)s %(thread)s %(message)s",level=logging.INFO)

def worker(event:Event,interval:int):
    while not event.wait(interval):
        logging.info("没有等到。。")

e = Event()
Thread(target=worker,args=(e,1)).start()

e.wait(5)
e.set()

print("======end========")

2.threading.Timer定时器，延迟执行

• threading.Timer继承自Thread,这个类用来定义延迟多久后执行一个函数。
• class threading.Timer(interval, function, args=None, kwargs=None)
  1. interval #多少时间后执行function函数
  2. function #需要执行的函数
• start方法执行之后，Timer对象会处于等待状态，等待了interval秒之后，开始执行function函数的。
• Timer是线程Thread的子类，Timer实例内部提供了一个finished属性，该属性是Event对象。
• cancel方法，本质上 是在worker函数执行前对finished属性set方法操作，从而跳过了worker函数执行，达到了取消的效果。

from threading import Timer
import logging
import time

logging.basicConfig(format="%(asctime)s %(threadName)s %(thread)s %(message)s",level=logging.INFO)

def worker():
    logging.info("in worker")
    time.sleep(5)
    logging.info("end in worker")

t = Timer(2,worker)
t.setName("timer1") #设置线程名称
# t.cancel() #取消定时器后，定时器不在执行
t.start()
# t.cancel() #取消定时器后，定时器不在执行
time.sleep(4) #等待4秒后，定时器已经开始执行
t.cancel() #当定时器执行后，无法取消

print("======end========")

3.threading.Lock锁

锁(Lock):一旦线程获得锁，其他试图获取锁的线程将被阻塞等待。
锁：凡是存在共享支援争抢的地方都可以使用锁，从而保证只有一个使用者可以完全使用这个资源。

import threading
import sys
import time

def print(*args):
    sys.stdout.write(" ".join(map(str,args))+"\n")

def worker(lock):
    print("worker start",threading.get_ident(),threading.current_thread().name)
    lock.acquire()
    print("worker over",threading.get_ident(),threading.current_thread().name)

lock = threading.Lock()
lock.acquire()
print(" -"*30)
for i in range(5):
    threading.Thread(target=worker,args=(lock,),name="w{}".format(i)).start()

print("- "* 30)
while True:
    time.sleep(0.1)
    cmd = input(">>").strip()
    if cmd == "r":
        lock.release()
    elif cmd == "q":
        break
    else:
        print(threading.enumerate())

上例可以看出不管在哪一个线程中，只要对一个已经上锁的锁阻塞请求，该线程就会阻塞。

• 加锁，解锁
  一般来说，加锁就需要解锁，但是加锁后解锁前，还要有一些代码执行，就有可能抛异常，一旦出现异常，锁是无 法释放，但是当前线程可能因为这个异常被终止了，这也产生了死锁

• 加锁、解锁常用语句：

  1. 使用try…finally语句保证锁的释放
  2. with上下文管理，锁对象支持上下文管理

• 计数器类，可加，可减。

import threading
import sys
import time

def print(*args):
    sys.stdout.write(" ".join(map(str,args))+"\n")

class Counter:
    def __init__(self):
        self._val = 0
        self.lock = threading.Lock()

    @property
    def value(self):
        with self.lock:
            return self._val

    def inc(self):
        with self.lock:
            self._val += 1

    def dec(self):
        with self.lock:
            self._val -= 1

def run(c:Counter,count=100):
    for _ in range(count):
        for i in range(-50,50):
            if i <0:
                c.dec()
            else:
                c.inc()

c = Counter()
c1 = 10 #线程数
c2 = 1000
for i in range(c1):
    threading.Thread(target=run,args=(c,c2)).start()

for k in threading.enumerate():
    if k.name != "MainThread":
        k.join()

print(c.value)

• 锁的应用场景
  锁适用于访问和修改同一个共享资源的时候，即读写同一个资源的时候。

• 使用锁的注意事项：

  1. 少用锁，必要时用锁。使用了锁，多线程访问被锁的资源时，就成了串行，要么排队执行，要么争抢执行
     • 举例，高速公路上车并行跑，可是到了省界只开放了一个收费口，过了这个口，车辆依然可以在多车道 上一起跑。过收费口的时候，如果排队一辆辆过，加不加锁一样效率相当，但是一旦出现争抢，就必须 加锁一辆辆过。注意，不管加不加锁，只要是一辆辆过，效率就下降了。
  2. 加锁时间越短越好，不需要就立即释放锁
  3. 一定要避免死锁

• 非阻塞锁的使用

import threading
import sys
import time

def print(*args):
    sys.stdout.write(" ".join(map(str,args))+"\n")

def worker(lock:threading.Lock):
    while True:
        if lock.acquire(False):
            print("do something.")
            time.sleep(1)
            lock.release()
            break
        else:
            print("try again")
            time.sleep(1)

lock = threading.Lock()
for i in range(5):
    threading.Thread(target=worker,name="w{}".format(i),args=(lock,)).start()

4.可重入锁RLock

• 可重入锁，是线程相关的锁
• 线程A获得可重复锁，并可以多次成功获取，不会阻塞。最后要在线程A中做和acquire次数相同的release

import threading
import sys
import time

def print(*args):
    sys.stdout.write(" ".join(map(str,args))+"\n")

def fib(num,rlock:threading.RLock):
    with rlock:
        if num<3:
            return 1
        return fib(num-1,rlock)+fib(num-2,rlock)

def work(num,rlock):
    print(fib(num,rlock))

rlock = threading.RLock()
for i in range(1,10):
    threading.Thread(target=work,args=(i,rlock)).start()

可重入锁
* 与线程相关，可在一个线程中获取锁，并可继续在同一线程中不阻塞多次获取锁
* 当锁未释放完，其它线程获取锁就会阻塞，直到当前持有锁的线程释放完锁
* 锁都应该使用完后释放。可重入锁也是锁，应该acquire多少次，就release多少次

5.Condition条件锁，等待通知

构造方法Condition(lock=None)，可以传入一个Lock或RLock对象，默认是RLock。

• 每个线程都可以通过Condition获取已把属于自己的锁，在锁中可以等待其他进程的同级锁的通知。当获取到同级锁的通知后，会停止等待。

• 当使用Condition(lock=Lock())初始化锁时，锁只能一级等待，不能出现多级等待。

• 简单示例：

import threading
import time

def work(cond:threading.Condition):
    with cond:
        print("开始等待")
        cond.wait()
        print("等到了")

cond = threading.Condition()
# cond = threading.Condition(threading.Lock())
threading.Thread(target=work,args=(cond,)).start()
threading.Thread(target=work,args=(cond,)).start()

with cond:
    with cond:
        time.sleep(1)
        print("开始释放二级等待")
        print(cond.notifyAll())
    time.sleep(2)
    print("开始释放一级等待")
    cond.notifyAll()

• 广播模式示例：

from threading import Thread,Condition,Lock
import time
import logging
import random

logging.basicConfig(format="%(asctime)s %(threadName)s %(thread)s %(message)s",level=logging.INFO)

class Dispachter:
    def __init__(self):
        self.data = None
        self.cond = Condition(lock=Lock())

    #生成者
    def produce(self,total):
        for _ in range(total):
            data = random.randint(1,100)
            with self.cond:
                logging.info("生产了一个数据：{}".format(data))
                self.data = data
                self.cond.notify(1)
            time.sleep(1) #模拟生成数据需要耗时1秒

    #消费者
    def consume(self):
        while True:
            with self.cond:
                self.cond.wait() #等待
                data = self.data
                logging.info("消费了一个数据 {}".format(data))
                self.data = None

d = Dispachter()
p = Thread(target=d.produce,name="producer",args=(10,))

# 增加消费者
for i in range(5):
    c = Thread(target=d.consume,name="consumer{}".format(i))
    c.start()

p.start()

上面例子中演示了生产者生产一个数据，就通知一个消费者消费。

• Condition总结
  Condition用于生产者消费者模型中，解决生产者消费者速度匹配的问题。采用了通知机制，非常有效率。
• 使用方式
  使用Condition，必须先acquire，用完了要release，因为内部使用了锁，默认使用RLock锁，最好的方式是使用 with上下文。
  消费者wait，等待通知。
  生产者生产好消息，对消费者发通知，可以使用notify或者notify_all方法。

6.therading.Semaphore信号量

和Lock很像，信号量对象内部维护一个倒计数器，每一次acquire都会减1，当acquire方法发现计数为0就阻塞请求 的线程，直到其它线程对信号量release后，计数大于0，恢复阻塞的线程。

• 注意：
  1. 计数器永远不会低于0，因为acquire的时候，发现是0，都会被阻塞。
  2. 信号量没有做超界限制

from threading import Semaphore

s =Semaphore(3)
print(s._value)
s.release() #会增加信号量
print(s._value) #可以看出没有做信号量上线控制
print("----------------")
print(s.acquire())
print(s._value)
print(s.acquire())
print(s._value)
print(s.acquire())
print(s._value)
print(s.acquire())
print(s._value)
print(s.acquire()) #当信号量为0再次acquire会被阻塞
print("~~~~~~阻塞了吗？")
print(s._value)

• 跨线程使用演示

from threading import Thread,Semaphore
import time
import logging

logging.basicConfig(format="%(asctime)s %(threadName)s %(thread)s %(message)s",level=logging.INFO)

#定义获取信号量
def worker(s:Semaphore):
    while s.acquire():
        logging.info("被执行了一次，获取一个信号量 _value={}".format(s._value))

#释放信号量
def cunn(s:Semaphore):
    while True:
        logging.info("释放一个信号量")
        s.release()
        time.sleep(1)

s = Semaphore(3)
#创建3个线程获取信号量
for i in range(3):
    Thread(target=worker,args=(s,),name="w{}".format(i)).start()

#开启一个线程释放信号量
Thread(target=cunn,args=(s,)).start()

7.threading.BoundedSemaphore有界信号量

• 有界信号量，不允许使用release超出初始值的范围，否则，抛出ValueError异常

from threading import BoundedSemaphore

bs = BoundedSemaphore(3)
print(bs._value)
bs.acquire()
bs.acquire()
bs.acquire()
print(bs._value)
bs.release()
bs.release()
bs.release()
print(bs._value)
bs.release()

• 应用举例
  连接池
  因为资源有限，且开启一个连接成本高，所以，使用连接池。
  一个简单的连接池
  连接池应该有容量（总数），有一个工厂方法可以获取连接，能够把不用的连接返回，供其他调用者使用。

from threading import Thread,BoundedSemaphore
import time
import logging
import random

logging.basicConfig(format="%(asctime)s %(threadName)s %(thread)s %(message)s",level=logging.INFO)

#链接类
class Conn:
    def __init__(self,name):
        self.name = name

class Pool:
    def __init__(self,count:int):
        self.count = count
        #池中放着链接备用
        self.pool = [self._connect("conn-{}".format(i)) for i in range(count)]
        self.bsemaphore = BoundedSemaphore(count)

    #创建连接方法，返回一个连接对象
    def _connect(self,conn_name):
        return Conn(conn_name)

    #获取一个链接
    def get_conn(self):
        self.bsemaphore.acquire()
        self.pool.pop()
        logging.info("从连接池拿走了一个连接~~~~~~~")

    #归还一个连接
    def return_conn(self,conn:Conn):
        logging.info("归还了一个连接----------")
        self.pool.append(conn)
        self.bsemaphore.release()

def worker(pool:Pool):
    conn = pool.get_conn()
    logging.info(conn)
    #模拟使用了一段时间
    time.sleep(random.randint(1,5))
    pool.return_conn(conn)

pool = Pool(3)
for i in range(6):
    Thread(target=worker,name="worker-{}".format(i),args=(pool,)).start()

上例中，使用信号量解决资源有限的问题。
如果池中有资源，请求者获取资源时信号量减1，拿走资源。当请求超过资源数，请求者只能等待。当使用者用完 归还资源后信号量加1，等待线程就可以被唤醒拿走资源。
注意：这个连接池的例子不能用到生成环境，只是为了说明信号量使用的例子，连接池还有很多未完成功能。

• 问题分析

1. 边界问题
   • 假设一种极端情况，计数器还差1就归还满了，有三个线程A、B、C都执行了第一句，都没有来得及release，这时 候轮到线程A release，正常的release，然后轮到线程C先release，一定出问题，超界了，直接抛异常。 因此信号量，可以保证，一定不能多归还。
2. 正常使用分析
   • 正常使用信号量，都会先获取信号量，然后用完归还。
   • 创建很多线程，都去获取信号量，没有获得信号量的线程都阻塞。能归还的线程都是前面获取到信号量的线程，其 他没有获得线程都阻塞着。非阻塞的线程append后才release，这时候等待的线程被唤醒，才能pop，也就是没有 获取信号量就不能pop，这是安全的。
   • 经过上面的分析，信号量比计算列表长度好，线程安全。

• 信号量和锁
  1. 信号量，可以多个线程访问共享资源，但这个共享资源数量有限。
  2. 锁，可以看做特殊的信号量，即信号量计数器初值为1。只允许同一个时间一个线程独占资源。

总结