引言:
线程之间经常需要协同工作,通过某种技术,让一个线程访问某些数据时,其它线程不能访问这些数据,直到该线程完成对数据的操作。这些技术包括临界区(critical section),互斥量(mutex),信号量(semaphore),事件event等。
event
threading库中的event对象通过使用内部一个flag标记,通过flag的true或者false的变化来进行操作。
名称 | 含义 |
set( ) | 标记设置为true |
clear( ) | 标记设置为false |
is_set( ) | 标记是否为true |
wait(timeout=none) | 设置等待标记为true的时长,none为无限等待。等到返回true,等不到返回false |
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
from threading import thread,event
import time
def creditor(event:event):
print ( "什么时候还我钱" )
event.wait()
print ( "我已经等了很长时间了" )
def debtor(event:event,count = 10 ):
print ( "可以宽裕几天吗?" )
money = []
while true:
print ( "先还你100" )
time.sleep( 0.5 )
money.append( 1 )
if len (money)>count:
event. set ()
break
print ( "我已经还完你的钱了" )
event = event()
c = thread(target = creditor,args = (event,))
d = thread(target = debtor,args = (event,))
c.start()
d.start()
|
运行结果如下所示:
可以看到creditor函数中因为event.wait( )线程进入等待状态,此时debtor线程进入运行,当满足条件时event.set( )将标记设置为true,creditor线程开始运行。谁wait就是等到flag变为true,或等到超时变为false。不限制等待的个数。
wait的使用
1
2
3
4
5
6
7
8
9
10
11
12
|
from threading import event,thread
def wait(event:event,interval):
while not event.wait(interval):
print ( "waiting for you" )
e = event()
thread(target = wait,args = (e, 3 )).start()
e.wait( 10 )
e. set ()
print ( "main exit" )
|
主线程一开始就wait 10s,waiting线程等待3s返回false,进入循环打印"waiting for you",重复3次,然后主线程set了,这时候waiting线程变为true,不再进入循环。
lock
凡是存在资源争用的地方都可以使用锁,从而保证只有一个使用者可以完全使用这个资源
现在要生产10个杯子,由10个工人开始生产
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
import threading
import time
cups = []
def worker(count = 10 ):
print ( "我是{},我开始生产了" . format (threading.current_thread().name))
flag = false
while true:
if len (cups)>count:
flag = true
time.sleep( 0.05 )
if not flag:
cups.append( 1 )
if flag:
break
print ( "finished.cups={}" . format ( len (cups)))
for _ in range ( 10 ):
threading.thread(target = worker,args = ( 1000 ,)).start()
|
运行结果如下图所示:
我们明明只需要到1000就会break,但是结果却到了1010个,这就是因为有10个线程,其中每个线程都在增加,但是增加后的数目,其他线程并不会知道(每个线程通过len函数拿到数量,但是刚拿到数字,其他线程就立即更新了)
这个时候我们就需要锁lock来实现了,一旦线程获得锁,其他试图获取锁的线程将被阻塞
名称 | 含义 |
acquire(blocking=true,timeout=-1) | 默认阻塞,阻塞可以设置超时时间。非阻塞时,timeout禁止设置。成功获取锁,返回true,否则返回false |
release( ) | 释放锁。可以从任何线程释放。已上锁的锁,会抛出runtimeerror异常 |
加锁的实现:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
import threading
import time
cups = []
lock = threading.lock()
def worker(count = 10 ):
print ( "我是{},我开始生产了" . format (threading.current_thread().name))
flag = false
while true:
lock.acquire()
if len (cups)> = count:
flag = true
time.sleep( 0.005 )
if not flag:
cups.append( 1 )
lock.release()
if flag:
break
print ( "finished,cups={}" . format ( len (cups)))
for _ in range ( 10 ):
threading.thread(target = worker,args = ( 1000 ,)).start()
|
运行结果如图所示:
一般来说加锁后还需要一些代码实现,在释放锁之前还有可能抛出异常,一旦出现异常,锁无法释放,但是当前这个线程会因为这个异常而终止,这样会产生死锁,因此使用时要使用如下的方法:
1,使用try...finally语句保证锁的释放
2,with安全上下文管理(锁对象支持上下文管理)
计数器类,用来加,减。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
import threading
import time
class counter:
def __init__( self ):
self ._val = 0
self .__lock = threading.lock()
@property
def value( self ):
return self ._val
def inc( self ):
try :
self .__lock.acquire()
self ._val + = 1
finally :
self .__lock.release()
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()
while true:
if threading.active_count() = = 1 :
print (c.value)
break
|
启动了10个线程,1000次从-50到50进行加减,最后得到0,如果没有加锁处理的话,得到的结果未必是自己想要的。
锁的使用场景:
锁适用于访问和修改同一个资源的时候,引起资源争用的情况下。使用锁的注意事项:
- 1,少用锁,除非有必要。多线程访问加锁的资源时,由于锁的存在,实际就变成了串行。
- 2,加锁时间越短越好,不需要就立即释放锁。
- 3,一定要避免死锁,使用with或者try...finally。
非阻塞锁使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
import threading
import time
def worker(tasks):
for task in tasks:
time.sleep( 0.001 )
if task.lock.acquire(false):
print ( "{} {} begin to start" . format (threading.current_thread(),task.name))
else :
print ( "{} {} is working" . format (threading.current_thread(),task.name))
class task:
def __init__( self ,name):
self .name = name
self .lock = threading.lock()
tasks = [task( 'task-{}' . format (x)) for x in range ( 10 )]
for i in range ( 5 ):
threading.thread(target = worker,name = "worker-{}" . format (i),args = (tasks,)).start()
|
运行结果如下图所示:
总共开启了5个线程,每个线程处理10个任务,因为在if语句里面,task.lock.acquire(false),所以每个线程只有拿到锁是true,其他的线程不会阻塞会返回false。打印"is working"。
以上所述是小编给大家介绍的python多线程之间的同步详解整合,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对服务器之家网站的支持!
原文链接:https://blog.51cto.com/berniem2m/2373692