本篇文章给大家带来的内容是关于Python中四种锁的使用示例(代码)，有一定的参考价值，有需要的朋友可以参考一下，希望对你有所帮助。

Lock互斥锁

使用前num = 0

def a():

global num

for _ in range(10000000):

num += 1

def b():

global num

for _ in range(10000000):

num += 1

if __name__ == '__main__':

t1=Thread(target=a)

t1.start()

t2=Thread(target=b)

t2.start()

t1.join()

t2.join()

print(num) #基本永远会小于20000000

使用后num = 0

def a(lock):

global num

for _ in range(1000000):

with lock:

num += 1

def b(lock):

global num

for _ in range(1000000):

with lock:

num += 1

if __name__ == '__main__':

lock = threading.Lock()

t1=Thread(target=a, args=(lock,))

t1.start()

t2=Thread(target=b, args=(lock,))

t2.start()

t1.join()

t2.join()

print(num) #永远会输出20000000

RLock重用锁#在之前的代码中永远不可能出现锁在没释放之前重新获得锁，但rlock可以做到，但只能发生在一个线程中，如：

num = 0

def a(lock):

with lock:

print("我是A")

b(lock)

def b(lock):

with lock:

print("我是b")

if __name__ == '__main__':

lock = threading.Lock()

t1 = Thread(target=a, args=(lock,))

t1.start() #会发生死锁，因为在第一次还没释放锁后，b就准备上锁，并阻止a释放锁

使用后if __name__ == '__main__':

lock = threading.RLock() #只需要改变锁为RLock程序马上恢复

t1 = Thread(target=a, args=(lock,))

t1.start()

Condition同步锁#这个程序我们模拟甲乙对话

Jlist = ["在吗", "干啥呢", "去玩儿不", "好吧"]

Ylist = ["在呀", "玩儿手机", "不去"]

def J(list):

for i in list:

print(i)

time.sleep(0.1)

def Y(list):

for i in list:

print(i)

time.sleep(0.1)

if __name__ == '__main__':

t1 = Thread(target=J, args=(Jlist,))

t1.start()

t1.join()

t2 = Thread(target=Y, args=(Ylist,))

t2.start()

t2.join() #上面的程序输出后发现效果就是咱们想要的，但是我们每次输出后都要等待0.1秒，也无法正好确定可以拿到时间片的最短时间值，并且不能保证每次正好都是另一个线程执行。因此，我们用以下方式，完美解决这些问题。

使用后Jlist = ["在吗", "干啥呢", "去玩儿不", "好吧"]

Ylist = ["在呀", "玩儿手机", "不去","哦"]

def J(cond, list):

for i in list:

with cond:

print(i)

cond.notify()

cond.wait()

def Y(cond, list):

for i in list:

with cond:

cond.wait()

print(i)

cond.notify()

if __name__ == '__main__':

cond = threading.Condition()

t1 = Thread(target=J, args=(cond, Jlist))

t2 = Thread(target=Y, args=(cond, Ylist))

t2.start()

t1.start() #一定保证t1启动在t2之后,因为notify发送的信号要被t2接受到，如果t1先启动，会发生阻塞。

Seamplore信号量

使用前class B(threading.Thread):

def __init__(self, name):

super().__init__()

self.name = name

def run(self):

time.sleep(1)

print(self.name)

class A(threading.Thread):

def __init__(self):

super().__init__()

def run(self):

for i in range(100):

b = B(i)

b.start()

if __name__ == '__main__':

a = A()

a.start() #执行后发现不断在输出

使用后class B(threading.Thread):

def __init__(self, name, sem):

super().__init__()

self.name = name

self.sem = sem

def run(self):

time.sleep(1)

print(self.name)

sem.release()

class A(threading.Thread):

def __init__(self, sem):

super().__init__()

self.sem = sem

def run(self):

for i in range(100):

self.sem.acquire()

b = B(i, self.sem)

b.start()

if __name__ == '__main__':

sem = threading.Semaphore(value=3)

a = A(sem)

a.start() #通过执行上面的代码，我们发现一次只能输出三个数字，sem控制访问并发量