更多优质内容
请关注公众号
请关注公众号
多线程编程 - 线程同步
什么是线程同步，为什么要线程同步
线程同步是为了解决多线程编程中，由于竞争使用资源或修改变量而造成数据不一致的问题举一个例子：
# coding=utf-8
import dis
a=0
def add():
global a
a+=1
return a
def desc():
global a
a-=1
return a
print(dis.dis(add))
print(dis.dis(desc))
10 0 LOAD_GLOBAL 0 (a)
2 LOAD_CONST 1 (1)
4 INPLACE_ADD
6 STORE_GLOBAL 0 (a)
12 8 LOAD_GLOBAL 0 (a)
10 RETURN_VALUE
16 0 LOAD_GLOBAL 0 (a)
2 LOAD_CONST 1 (1)
4 INPLACE_SUBTRACT
6 STORE_GLOBAL 0 (a)
18 8 LOAD_GLOBAL 0 (a)
10 RETURN_VALUE
我们查看一下 add() 和 desc() 这两个方法的字节码。
可以看出 add() 中 a+=1 可以由4行字节码组成：
1. 将变量a加载到内存
2. 将常量1加载到内存
3. 在内存中执行a+1操作
4. 将内存中的中间值a+1赋值给代码中的变量a
desc()同样也有这4步
1. 将变量a加载到内存
2. 将常量1加载到内存
3. 在内存中执行a-1操作
4. 将内存中的中间值a-1赋值给代码中的变量a
python在实际运行的时候是一行行字节码运行的。
如果上面将add()和desc()分为两个线程循环运行，就可能会出现这种情况：
一开始a=0,add()运行完第3行字节码的时候，时间片用完，内存中的中间值是a+1=1。轮到desc()执行，并且desc()执行完了4行字节码，此时脚本中a=-1。又轮到add()运行，执行第4行字节码，将1赋给脚本变量a,得到a=1
所有 add()和desc()各运行1次，应该a=0才对，实际上a=1或者a=-1。导致数据不一致。
而造成这个问题的根本原因就是：字节码1~4不是一个原子操作。
线程同步可以保证 a+=1和a-=1内部的字节码的运行是一个原子操作，从而避免数据被改乱
同步方式1：Lock互斥锁 和 RLock重入锁
Lock互斥锁
互斥锁可以保证锁内的竞争资源同一时刻只能被一个线程使用。一个线程想使用某一竞争资源，就要先获取锁。如果资源已被线程B加了锁，那么线程A想获取锁使用资源的时候就会进入休眠，等待线程B释放锁。
底层的细节就是，线程A会进入休眠，主动让出CPU给线程B，让线程B执行完锁内的代码然后释放锁，线程A才会被唤醒，并且拿到锁对资源操作。
如果有多个线程因为等待锁而进入休眠，这些线程会被放到一个等待队列中。先进队列的线程可以先被唤醒而获取锁和资源的使用。
线程加了互斥锁，就必须要释放锁，否则其他线程会一直等待。
锁的优点：
可以实现线程间同步，保证线程安全，资源的有序使用，变量不会被改乱。
锁的缺点：
1.加锁和释放锁会消耗时间，所以锁会影响程序性能
2.锁可能会引起死锁
例子1：互斥锁的基本使用
使用互斥锁的经典例子：
# coding=utf-8
from threading import Thread,Lock
a=0
lock = Lock() # 定义一个互斥锁
def add(lock):
global a
for i in range(1000000):
lock.acquire() # 加锁
a+=1
lock.release() # 释放锁
def desc(lock):
global a
for i in range(1000000):
lock.acquire() # 加锁
a-=1
lock.release() # 释放锁
if __name__=="__main__":
t1 = Thread(target=add,args=(lock,))
t2 = Thread(target=desc,args=(lock,))
t1.start()
t2.start()
t1.join()
t2.join()
print(a)
PS：
1、如果不加锁，得到的a不是0，说明加锁可以保证操作的原子性，保护资源不被改乱
2、如果将加锁和释放锁放在for循环之外，此时一个线程会等待另一个线程完全执行完for才会能执行，相当于变成单线程，这是错误的使用锁的方式。
例子2：模拟死锁的两种情况
a.死锁的第一种情况，锁嵌套（在锁中嵌套的获取同一把锁）
from threading import Lock
class ThreadSafeQueue:
def __init__(self):
self.queue=[] # 用列表模拟一个队列，使用锁保证queue是线程安全的
self.lock = Lock() # 定义一个互斥锁
def size(self):
self.lock.acquire()
size = len(self.queue)
self.lock.release()
def get(self):
item = None
self.lock.acquire()
if self.size()>0:
item = self.queue.pop(0)
self.lock.release()
def put(self,item):
self.lock.acquire()
self.queue.append(item)
self.lock.release()
if __name__=="__main__":
q = ThreadSafeQueue()
q.put(1)
q.put(2)
print(q.get()) # 发生一直阻塞
问题其实出在：
def get(self):
item = None
self.lock.acquire()
if self.size()>0:
item = self.queue.pop(0)
self.lock.release()
self.size()中加了锁，在if self.size>0: 外面又加了相同的锁，所以相同的锁嵌套，就会造成里面的锁等待外面的锁释放。
但是释放锁的代码在内层锁的代码之后，而执行到内层锁的代码时就已经阻塞了，因此永远都不肯执行释放锁的代码。
结果一直阻塞。
简单的说就是：
lock1.acquire()
lock1.acquire() # 该行发生死锁，一直阻塞
do_something()
lock1.release()
lock1.release()
这种情况肯定会死锁
b.死锁的第二种情况，相互等待资源的锁释放。
假如有线程A,B，资源m,n，锁X,Y 。线程A，B都会用到资源m,n，使用资源m要用锁X保护，使用资源n要用锁Y保护
A的逻辑要求先操作m再操作n，而且m和n的操作具有原子性
B的逻辑要求先操作n再操作m，而且m和n的操作具有原子性
# coding=utf-8
from threading import Thread,Lock
lock_X = Lock()
lock_Y = Lock()
m = 0
n = 0
def task1():
global m,n,lock_X,lock_Y
for i in range(100000):
lock_X.acquire()
m+=1 # 1
print("task1 add m")
lock_Y.acquire() # 2
n+=1
print("task1 add n")
lock_Y.release()
lock_X.release()
def task2():
global m,n,lock_X,lock_Y
for i in range(100000):
lock_Y.acquire()
n-=1 # 3
print("task2 desc n")
lock_X.acquire() # 4
m-=1
print("task2 desc m")
lock_X.release()
lock_Y.release()
if __name__=="__main__":
t1 = Thread(target=task1)
t2 = Thread(target=task2)
t1.start()
t2.start()
过程： task1运行到#1时间片用完，接着CPU切换到task2，task2运行完#3，现在要运行#4，但是锁X被task1获取所以task2要等待锁X释放。所以CPU切换会task1，从上一次#1的地方继续执行，执行到 #2发现锁Y已经被task2获取，所以等待task1释放锁Y。
结果：task1等待task2释放锁Y,task2等待task1释放锁X。双方相互等待，发生死锁。
情况a和情况b的区别：a是同一把锁嵌套，锁等待自己这把锁造成死锁；b是两把不同的锁嵌套，造成相互等待造成死锁
例子3：多线程的有序进行
# coding=utf-8
lock1 = Lock()
lock2 = Lock()
lock3 = Lock()
lock2.acquire()
lock3.acquire()
def task1():
while True:
lock1.acquire()
print("task1")
sleep(0.5)
lock2.release()
def task2():
while True:
lock2.acquire()
print("task2")
sleep(0.5)
lock3.release()
def task3():
while True:
lock3.acquire()
print("task3")
sleep(0.5)
lock1.release()
if __name__=="__main__":
t1 = Thread(target=task1)
t2 = Thread(target=task2)
t3 = Thread(target=task3)
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()
这个例子的有趣之处在于
1.三个线程并发，但同一时间只有一把锁是处于释放状态，其他两个锁是处于锁定状态。
2.输出数据时会加锁，输出完之后却会释放另一个线程要获取的锁而不是释放自己的这把锁。这意味着，这个线程执行完之后，只有要获取“上一个线程释放了的锁”的线程能执行，其他线程不能执行（因为这些线程想获取的锁被其他线程给锁住了）。通过之中方式可以指定线程的执行顺序。
上面的线程是没有实际意义的，因为三个线程相当于是串行执行的，相比于单线程不用切换不用加锁和释放锁，这样串行的多线程效率反而比单线程还低些
例子4：
我们的目的是这样的：
有4个列表list1~4,list1包含初始元素，list2~4是空的。
我希望对list1中的数据x进行以下运算 (x+1)*2-3
元素从list1弹出，+1，再放到list2，使用线程1完成 |
元素从list2弹出，*2，再放入list3，使用线程2完成 |--这三个过程是（并发的）同时发生的
元素从list3弹出，-3，再放入list4，使用线程3完成 |
下面我们计划一下，哪些过程需要加锁，哪些过程不用加锁
互斥锁是为了解决资源竞争，所以我们需要对线程共享的资源加锁即可。
list1只有线程1用到了，list4只有线程3用到了，所以对这两个list的操作不用加锁
list2会被线程1和线程2两个线程用到，所以线程1,2在操作list2时要加锁
list3同理，线程2,3操作list3时需要加锁。
list1~3取出来的元素需要进行算数运算，而算术运算是由一个线程单独完成，所以每个元素运算的时候只会被1个线程使用，所以元素进行算数运算的过程无需加锁。
# coding=utf-8
from threading import Thread,Lock
from time import sleep,time
lock1 = Lock()
lock2 = Lock()
list2=[]
list3=[]
list4=[]
# 将元素从list1取出，进行+1处理，放入list2
def task1():
while len(list1):
item = list1.pop() # 取出元素
item+=1
# 放入list2时要对list2上锁
lock1.acquire()
list2.append(item)
print("task1 append %d" % item)
lock1.release()
# 将元素从list2取出，进行*2处理，放入list3
def task2():
while True:
if len(list2):
# 从list2取出元素时要对list2上锁
lock1.acquire()
item = list2.pop()
print("task2 pop %d" % item)
lock1.release()
item=item*2 # 这里不用放到锁内
# 将元素放入list3要对list3上锁，而且因为list2和list3是两个不同的资源，所以要用另一个锁来锁住list3
lock2.acquire()
list3.append(item)
print("task2 append %d" % item)
lock2.release()
else:
sleep(0.0001)
# 将元素从list3取出，进行-3处理，放入list4
def task3():
while True:
if len(list3):
lock2.acquire()
item = list3.pop()
print("task3 pop %d" % item)
lock2.release()
item-=3 # 这句不用放到锁内
list4.append(item)
print("task3 append %d" % item)
else:
sleep(0.0001)
if __name__=="__main__":
list1 = list(range(100000))
t1 = Thread(target=task1)
t2 = Thread(target=task2)
t3 = Thread(target=task3)
t1.start()
t2.start()
t3.start()
t1.join()
t2.join()
t3.join()
在使用互斥锁的时候，多线程中加了锁的那段代码是串行的。其实在写程序的时候，我总是会想整个运行过程哪些过程是可以同时发生的（并发的），哪些过程是串行的，因为可以同时执行的地方就是多线程比单线程提升了效率的地方，而串行的地方其效率和单线程一样的。
其实我们只用知道哪些地方是没有并发的就可以了，知道了哪些地方是没有并发的，那么其他所有地方都是并发的
例如：我想知道 元素A压入list2，已经从list2弹出的元素B压入list3，已经从list3弹出的元素C压入list4 这三个动作是可以同时发生的吗？
可以。因为从上面加锁的地方知道：元素A压入list2和元素B从list2弹出是不能同时进行的，元素B压入list3和元素C弹出list3不能同时进行。其他操作都可以同时发生。
总结：程序中只需对多个线程共享的资源进行上锁，对独享的资源无需上锁
RLock 重入锁
重入锁的特点：
可以允许同一个锁进行嵌套，但是释放锁的次数一定要等于获取锁的次数。
一个线程内的重入锁可以嵌套而不阻塞，但是线程间使用同一个重入锁就和互斥锁一样会阻塞
例如 将死锁情况a中的Lock换成RLock重入锁，程序就不会死锁
from threading import RLock
class ThreadSafeQueue:
def __init__(self):
self.queue=[] # 用列表模拟一个队列，使用锁保证queue是线程安全的
self.lock = RLock() # 定义一个互斥锁
def size(self):
self.lock.acquire()
size = len(self.queue)
self.lock.release()
return size
def get(self):
item = None
self.lock.acquire()
if self.size()>0:
item = self.queue.pop(0)
self.lock.release()
return item
def put(self,item):
self.lock.acquire()
self.queue.append(item)
self.lock.release()
if __name__=="__main__":
q = ThreadSafeQueue()
q.put(1)
q.put(2)
print(q.get())
print(q.get())
print(q.get())
下面贴出RLock的源码
class _RLock:
def __init__(self):
self._block = _allocate_lock() # 获取一把互斥锁
self._owner = None # 用于记录RLock调用acquire()时的线程，是一个整数
self._count = 0 # 记录某一个线程调用了几次 RLock.acquire()
def acquire(self, blocking=True, timeout=-1):
me = get_ident() # 获取当前线程的唯一标识，是一个整数
if self._owner == me: # 如果是本线程第二次以上的调用 RLock.acquire() 那么只把计数器+1，但是不会调用互斥锁self._block的acquire()；但如果是别的线程第二次以上的调用RLock.acquire() 那么就会再调用一次互斥锁self._block的acquire(),该线程进入休眠等待状态
self._count += 1
return 1
rc = self._block.acquire(blocking, timeout) # 线程第一次调用 RLock.acquire()会调用互斥锁的 acquire()
if rc: # # 线程第一次调用 RLock.acquire()会进入该条件
self._owner = me
self._count = 1
return rc
__enter__ = acquire
def release(self):
if self._owner != get_ident():
raise RuntimeError("cannot release un-acquired lock")
self._count = count = self._count - 1 # 每调用一次RLock.release()就会将计数器-1
if not count: # 只有当调用 RLock.release()的次数等于RLock.acquire()的次数才会真正调用 self._block的release()释放锁。
self._owner = None
self._block.release()
总结： RLock是通过互斥锁和计数器实现的
