各种锁

Posted 2020-11-28 saoqiang

死锁现象

第一种 加了2次同样的锁 只存在于互斥锁

from threading import Thread
from threading import Lock
def task(lock):
    lock.acquire()
    lock.acquire()
    print('66666')
    lock.release()
    lock.release()
if __name__ == '__main__':
    lock = Lock()
    t=Thread(target=task,args=((lock,)))
    t.start()

第2种 2个进程都想获取对方的锁 却不可能实现

from threading import Thread
from threading import Lock
import time
def fun1(lock_A,lock_B):
    lock_A.acquire()
    print(f'{t.name}拿到了A锁')
    lock_B.acquire()
    print(f'{t.name}拿到了B锁')
    lock_B.release()
    lock_A.release()
def fun2(lock_A,lock_B):
    lock_B.acquire()
    print(f'{t.name}拿到了B锁')
    time.sleep(0.1)
    lock_A.acquire()
    print(f'{t.name}拿到了A锁')
    lock_A.release()
    lock_B.release()
def task(lock_A,lock_B):
    fun1(lock_A,lock_B)
    fun2(lock_A,lock_B)
if __name__ == '__main__':
    lock_A = Lock()
    lock_B = Lock()
    for i in range(3):
        t=Thread(target=task,args=((lock_A,lock_B)))
        t.start()

可重复锁RLock

可重复锁，是线程相关的锁不管实列化多少次都是同一只把锁,

引用计数,只要计数不为0,其他线程不可以抢.

可以解决死锁现象

from threading import Thread
from threading import RLock
from multiprocessing import RLock
import time
def fun1(lock_A,lock_B):
    lock_A.acquire()
    print(f'{t.name}拿到了A锁')
    lock_B.acquire()
    print(f'{t.name}拿到了B锁')
    lock_B.release()
    lock_A.release()
def fun2(lock_A,lock_B):
    lock_B.acquire()
    print(f'{t.name}拿到了B锁')
    time.sleep(0.1)
    lock_A.acquire()
    print(f'{t.name}拿到了A锁')
    lock_A.release()
    lock_B.release()
def task(lock_A,lock_B):
    fun1(lock_A,lock_B)
    fun2(lock_A,lock_B)
if __name__ == '__main__':
    lock_B=lock_A = RLock()
    for i in range(5):
        t=Thread(target=task,args=((lock_A,lock_B)))
        t.start()

信号量Semaphore

本质就是一个计数器，用来为多个进程共享的数据结构提供受控访问。
控制并发数量

from threading import Semaphore
from threading import Thread
from threading import current_thread
import time
import random

def task(sem):
    sem.acquire()
    print(f'{current_thread().name}在运行中')
    time.sleep(random.randint(1,3))
    sem.release()

if __name__ == '__main__':
    sem=Semaphore(4)
    for i in range(50):
        t=Thread(target=task,args=((sem,)))
        t.start()

GIL全局解释器锁

GIL锁: 全局解释器锁. Cpython特有的一把互斥锁,自动加锁解锁将并发变成串行
同一时刻同一进程中只有一个线程被执行使用共享资源 ,牺牲效率,保证数据安全.
为什么加锁?
当时都是单核时代,而且cpu价格非常贵.
如果不加全局解释器锁, 开发Cpython解释器的程序员就会在源码内部各种主动加锁,解锁,非常麻烦,各种死锁现象等等.他为了省事儿,直接进入解释器时给线程加一个锁.

优点: 保证了Cpython解释器的数据资源的安全.
缺点: 单个进程的多线程不能利用多核.
Jpython没有GIL锁.
pypy也没有GIL锁.

GIL与lock锁的区别

相同点: 都是同种锁,互斥锁.
不同点:
GIL锁全局解释器锁,保护解释器内部的资源数据的安全.
GIL锁 上锁,释放无需手动操作. 只有遇到io阻塞就会释放
自己代码中定义的互斥锁保护进程中的资源数据的安全.
自己定义的互斥锁必须自己手动上锁,释放锁.

验证计算密集型IO密集型的效率

不能用多核会影响效率吗？

看处理数据情况

我们有四个任务需要处理，处理方式肯定是要玩出并发的效果，解决方案可以是：
方案一：开启四个进程
方案二：一个进程下，开启四个线程
#单核情况下，分析结果: 
　　如果四个任务是计算密集型，没有多核来并行计算，方案一徒增了创建进程的开销，方案二胜
　　如果四个任务是I/O密集型，方案一创建进程的开销大，且进程的切换速度远不如线程，方案二胜

#多核情况下，分析结果：
　　如果四个任务是计算密集型，多核意味着并行计算，在python中一个进程中同一时刻只有一个线程执行用不上多核，方案一胜
　　如果四个任务是I/O密集型，再多的核也解决不了I/O问题，方案二胜

总结:
多核的前提下: 如果任务Io密集型: 多线程并发.
如果任务计算密集型: 多进程并发

验证纯计算型实列

from multiprocessing import Process
from threading import Thread
import time
# def task():
#     count=0
#     for i in range(100000000):
#         count += 1
# if __name__ == '__main__':
#     time_1=time.time()
#     l1=[]
#     for i in range(5):
#         p=Process(target=task,args=(()))
#         p.start()
#         l1.append(p)
#     for i in l1:
#         i.join()
#     print(f'总用时{time.time()-time_1}')#14.197925329208374


from threading import Thread
import time
def task():
    count=0
    for i in range(100000000):
        count += 1
if __name__ == '__main__':
    time_2=time.time()
    l1=[]
    for i in range(5):
        p=Thread(target=task,args=(()))
        p.start()
        l1.append(p)
    for i in l1:
        i.join()
    print(f'总用时{time.time()-time_2}')#25.88302206993103

验证io阻塞型实列

from multiprocessing import Process
from threading import Thread
import time
def task():
    count=0
    time.sleep(1)
    for i in range(666666):
        count += 1
if __name__ == '__main__':
    time_1=time.time()
    l1=[]
    for i in range(5):
        p=Process(target=task,args=(()))
        p.start()
        l1.append(p)
    for i in l1:
        i.join()
    print(f'总用时{time.time()-time_1}')#1.4887681007385254


# from threading import Thread
# import time
# def task():
#     count=0
#     time.sleep(1)
#     for i in range(100):
#         count += 1
# if __name__ == '__main__':
#     time_2=time.time()
#     l1=[]
#     for i in range(5):
#         p=Thread(target=task,args=(()))
#         p.start()
#         l1.append(p)
#     for i in l1:
#         i.join()
#     print(f'总用时{time.time()-time_2}')#1.0018517971038818