线程同步:
线程间协同,通过某种技术,让一个线程访问某些数据时,其它线程不能访问这些数据,直到该线程完成对数据的操作;
结果要能预知;
critical section,临界区;
mutex,互斥量,threading.Lock类; #即锁,你用我不用,我用你不用
semaphore,信号量;
event,事件,threading.Event类;
barrier,屏障、栅栏,threading.Barrier;
帮助:
Python 3.5.3 documentation-->Library Reference-->Concurrent Execution-->threading
目录
threading.Event类:... 1
theading.Lock类:... 4
阻塞锁:... 5
非阻塞锁:... 10
event,事件,是线程间通信机制中最简单的实现,使用一个内部的flag标记,通过flag的True或False的变化来进行操作;
e.set(),标记设置为True;
e.clear(),标记设置为False;
e.is_set(),标记是否为True;
e.wait(timeout=None),Block until the internal flag is true,设置等待标记为True的时长,None为无限等待,等到返回True,未等到超时了返回False;
e.wait()的使用:
wait优于sleep,wait会主动让出时间片,其它线程可以被调度,而sleep会占用时间片不让出;
例:
def do(interval, e: threading.Event):
while not e.wait(interval):
logging.info('do sth')
e = threading.Event()
threading.Thread(target=do, args=(3,e)).start()
e.wait(5) #可用time.sleep(5),e.wait(5)优于time.sleep(5)
e.set()
print('main exit')
输出:
2018-07-30-14:58:08 Thread info: 10268 Thread-1 do sth
main exit
例:
老板雇佣了一个工人,让他生产杯子,老板一直等着工人,直到生产了10个杯子;
import threading
import time
import logging
FORMAT = '%(asctime)-15s\tThread info: %(thread)d %(threadName)s %(message)s'
logging.basicConfig(level=logging.INFO, format=FORMAT, datefmt='%Y-%m-%d-%H:%M:%S')
cups = [] #当前,如果多个工人做则有问题
event = threading.Event() #实例,主线程、boss、worker都要看event,注意其作用域,使用同一个Event对象的flag
def boss(e: threading.Event):
logging.info("I'm boss,waiting for U.")
e.wait()
logging.info('good job.')
def worker(n, e: threading.Event):
while True:
time.sleep(0.5)
cups.append(1)
logging.info('make 1')
if len(cups) >= n:
# logging.info('I finished my job. {}'.format(len(cups))) #此处打印也可放到下面,老板在主线程中等待,if event.wait():中
e.set()
break
logging.info('I finished my job. {}'.format(len(cups)))
b = threading.Thread(target=boss, args=(event,))
w = threading.Thread(target=worker, args=(10, event))
w.start()
b.start()
# if event.wait(): #方1 ,老板在主线程中等待
# logging.info('I finished my job. {}'.format(len(cups)))
# while not event.wait(1): #方2,1秒看1次,老板在主线程中等待
# pass
# logging.info('I finished my job. {}'.format(len(cups)))
# while not event.is_set(): #方3,老板在主线程中等待
# event.wait() #谁wait就是等到flag变为True,或等到超时返回False,不限制等待的个数
# logging.info('I finished my job. {}'.format(len(cups)))
输出:
2018-07-30-14:47:12 Thread info: 11032 Thread-1 I'm boss,waiting for U.
2018-07-30-14:47:13 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:13 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:14 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:14 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:15 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:15 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:16 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:16 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:17 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:17 Thread info: 6996 Thread-2 make 1
2018-07-30-14:47:17 Thread info: 6996 Thread-2 I finished my job. 10
2018-07-30-14:47:17 Thread info: 11032 Thread-1 good job.
例:
Event练习:
实现Timer,延时执行的线程;
延时计算add(x,y);
思路:
Timer的构造函数中参数得有哪些?
如何实现start启动一个线程执行函数?
如何cancel取消待执行任务?
class Timer:
def __init__(self, interval, fn, *args, **kwargs):
self.interval = interval
self.fn = fn
self.args = args
self.kwargs = kwargs
self.event = threading.Event()
def start(self):
threading.Thread(target=self.__do).start()
def cancel(self):
self.event.set()
def __do(self):
self.event.wait(self.interval) #等待超时后返回False
# print(self.event.__dict__) # {'_flag': False, '_cond': <Condition(<unlocked _thread.lock object at 0x00000000012435D0>, 0)>}
print(self.event.is_set()) #False
if not self.event.is_set():
self.fn(*self.args, **self.kwargs)
def add(x, y):
logging.info(x+y)
t = Timer(3, add, 4, 5)
t.start()
#t.cancel() #有此句self.is_set()为True
输出:
False
2018-07-31-09:29:45 Thread info: 5156 Thread-1 9
解决多线程中数据同步;
凡是存在共享资源争抢的地方都可使用锁,从而保证只有一个使用者可完全使用这个资源;
程序=数据+算法,要保证数据的结果是可预知的,若不可预知该程序是没用的,为得到正确的结果,性能是其次已不重要了;
一旦某一线程获得锁,其它试图获取的线程将被阻塞;
acquire(blocking=True,timeout=-1),默认阻塞,阻塞可设置超时时间;非阻塞时,timeout禁止设置;成功获取锁,返回True,否则返回False;
release(),释放锁,可从任何线程调用释放;该方法执行后,已上锁的锁会被重置为unlocked;在未上锁的锁上调用release(),抛RuntimeError: release unlocked lock;
注:
锁将acquire()和release()中间的代码管住;
大多数情况下用的是阻塞锁;
锁,类似ATM机的小房子;
加锁,应在分析业务基础之上,在合适的地方加,不能从前到后加一把大锁,这样效率低下;
加锁、解锁:
一般来说,加锁后还要有一些代码实现,在释放锁之前还有可能抛异常,一旦出现异常,锁无法释放,但当前线程可能因为这个异常终止了,这就产生了死锁;
一般谁(某个线程)加的锁谁解锁;
常用语句:
try...finally,保证锁的释放;
with,Lock类中有__enter__()和__exit__(),锁对象支持上下文管理;
锁的应用场景:
锁,适用于访问和修改同一个共享资源的时候,即读写同一个资源的时候;
如果全部都是读取同一个共享资源需要锁吗?不需要,因为这时可认为共享资源是不可变的,每一次读取它都是一样的值,所以不用加锁;
不使用锁,虽有了效率,但结果是错的;
使用了锁,虽效率低下,但结果是对的,所以为了对的结果,让计算机计算去吧!
使用锁的注意事项:
少用锁,必要时加锁,使用了锁,多线程访问被锁的资源时,就成了串行,要么排队执行,要么争抢执行;
例如,高速公路上的车并行跑,到了省界只开放了一个收费口,过了这个口,车辆依然可在多车道上并行跑;过收费口时,如果排队一辆一辆过,加不加锁一样,效率相当,但一旦出现争抢,就必须加锁一辆辆过;
加锁时间越短越好,不需要就立即释放锁;
一定要避免死锁,尽量用with语句;
lock.acquire()
例:
lock = threading.Lock()
lock.acquire()
print('get locker1')
lock.acquire() #block,再次获取同一个锁时阻塞(独占),必须等别人将该锁释放才能获取到,此行之后的语句不能执行;即,某一线程获得锁后,另一线程只要执行到lock.acquire()就会阻塞住(卡住)
print('get locker2')
lock.release() #一旦某一线程release,等着的线程争抢,谁抢到谁上锁;一般情况,有多少个acquire就有多少个release;要保证一定release,用到上下文(with语句)或try...finally
print('release locker')
注:
死锁(解不开的锁,就是死锁):
同一个锁,你等我释放、我等你释放、自己等自己释放;
两个锁,你等我放一把,我等你放一把,互相等待,都不释放,都为阻塞状态;
例:
def work():
time.sleep(3)
lock.release() #任何一个线程都可对同一个锁操作,此处释放锁
lock = threading.Lock()
lock.acquire()
print('get locker1')
threading.Thread(target=work).start()
time.sleep(5)
lock.acquire()
print('get locker2')
lock.release()
print('release locker')
例:
需求:10个工人完成100个杯子;
注:临界点问题:
当生产到99个时,若每个线程都可看到,每个线程看到转头去生产了,导致结果会多;
只允许一个线程可看到当前生产杯子的数量,其它线程不可见,这个线程看到后生产完1个杯子时,其它线程均可看到,杯子数量达到要求就不生产了;
cups = []
def worker(task=100):
while True:
count = len(cups)
logging.info('current count: {}'.format(count))
if count >= task:
break
cups.append(1)
logging.info('{} make 1'.format(threading.current_thread()))
logging.info('total: {}'.format(count))
for _ in range(10):
threading.Thread(target=worker).start()
输出:
……
2018-08-06-09:13:14 Thread info: 10208 Thread-2 current count: 104
2018-08-06-09:13:14 Thread info: 10208 Thread-2 total: 104
2018-08-06-09:13:14 Thread info: 10252 Thread-6 <Thread(Thread-6, started 10252)> make 1
2018-08-06-09:13:14 Thread info: 10252 Thread-6 current count: 104
2018-08-06-09:13:14 Thread info: 10252 Thread-6 total: 104
解决,加锁:
cups = []
lock = threading.Lock()
def worker(lock:threading.Lock, task=100):
while True:
lock.acquire()
count = len(cups)
logging.info('current count: {}'.format(count))
lock.release()
if count >= task: #虽不是大锁,仍有问题,在临界点时,其它线程仍可看到杯子总数,继而再做;解决,把中间代码全部放到锁里
break
lock.acquire()
cups.append(1)
lock.release()
logging.info('{} make 1'.format(threading.current_thread()))
logging.info('total: {}'.format(count))
for _ in range(10):
threading.Thread(target=worker, args=(lock,)).start()
输出:
……
2018-08-06-09:19:11 Thread info: 4092 Thread-4 current count: 100
2018-08-06-09:19:11 Thread info: 4092 Thread-4 total: 100
2018-08-06-09:19:11 Thread info: 11024 Thread-6 current count: 100
2018-08-06-09:19:11 Thread info: 11024 Thread-6 total: 100
例:
cups = []
lock = threading.Lock()
def worker(lock:threading.Lock, task=100):
while True:
lock.acquire() #该锁为互斥锁,你有我没有,我有你没有
count = len(cups)
logging.info('current count: {}'.format(count))
# lock.release()
if count >= task:
lock.release()
break #仍有问题,break后未释放锁,解决try...finally或上下文
# lock.acquire()
cups.append(1)
lock.release()
logging.info('{} make 1'.format(threading.current_thread()))
logging.info('total: {}'.format(count))
for _ in range(10):
threading.Thread(target=worker, args=(lock,)).start()
输出:
……
2018-08-06-09:23:38 Thread info: 10148 Thread-5 current count: 99
2018-08-06-09:23:38 Thread info: 10148 Thread-5 <Thread(Thread-5, started 10148)> make 1
2018-08-06-09:23:38 Thread info: 10148 Thread-5 current count: 100
2018-08-06-09:23:38 Thread info: 10148 Thread-5 total: 100
例:
计数器类,可以加、可以减;
class Counter:
def __init__(self):
self.__val = 0
def inc(self):
self.__val += 1
def dec(self):
self.__val -= 1
@property
def value(self):
return self.__val
def do(c:Counter, count=100):
for _ in range(count):
for i in range(-50,50):
if i < 0:
c.dec()
else:
c.inc()
c = Counter()
threadcount = 10 #依次10,100,1000查看,结果没变化
c2 = 1000 #依次10,100,1000查看,当1000时,每个线程耗cpu时间长,结果将不可预知
for i in range(threadcount):
threading.Thread(target=do, args=(c,c2)).start()
# time.sleep(5)
# print(c.value) #多线程情况下,每个线程耗cpu时间长的情况下,此处打印的值不是最终结果
while True:
time.sleep(3)
print(threading.enumerate())
print(c.value)
输出:
[<_MainThread(MainThread, started 7424)>]
81
[<_MainThread(MainThread, started 7424)>]
81
例,解决,加锁:
class Counter:
def __init__(self):
self.__val = 0
self.__lock = threading.Lock()
def inc(self):
try: #同with self.__lock: self.__val += 1;
self.__lock.acquire()
self.__val += 1
finally:
self.__lock.release()
def dec(self):
with self.__lock:
self.__val -= 1
@property
def value(self):
with self.__lock:
return self.__val
def do(c:Counter, count=100):
for _ in range(count):
for i in range(-50,50):
if i < 0:
c.dec()
else:
c.inc()
c = Counter()
threadcount = 10
c2 = 1000
for i in range(threadcount):
threading.Thread(target=do, args=(c,c2)).start()
# time.sleep(5)
# print(c.value)
while True:
time.sleep(1)
# print(threading.enumerate())
# print(c.value)
if threading.active_count() == 1:
print(threading.enumerate())
print(c.value)
else:
print(threading.enumerate())
输出:
[<Thread(Thread-2, started 11888)>, <Thread(Thread-3, started 8312)>, <Thread(Thread-6, started 10004)>, <_MainThread(MainThread, started 12200)>, <Thread(Thread-8, started 12012)>, <Thread(Thread-9, started 7788)>, <Thread(Thread-5, started 12032)>, <Thread(Thread-10, started 11576)>]
[<_MainThread(MainThread, started 12200)>]
0
[<_MainThread(MainThread, started 12200)>]
0
lock.acquire(False),获取到锁返回True,否则返回False;
例:
lock = threading.Lock()
lock.acquire()
ret = lock.acquire(False) #获取到锁则返回True
print(ret)
输出:
False
例:
cups = []
lock = threading.Lock()
def worker(lock:threading.Lock, task=100):
while True:
if lock.acquire(False):
count = len(cups)
logging.info('current count: {}'.format(count))
if count >= task:
lock.release()
break
cups.append(1)
lock.release()
logging.info('{} make 1'.format(threading.current_thread()))
logging.info('total: {}'.format(count))
for _ in range(10):
threading.Thread(target=worker, args=(lock,)).start()
输出:
……
2018-08-06-13:52:35 Thread info: 11232 Thread-9 current count: 100
2018-08-06-13:52:35 Thread info: 11232 Thread-9 total: 100
2018-08-06-13:52:35 Thread info: 11752 Thread-10 current count: 100
2018-08-06-13:52:35 Thread info: 11752 Thread-10 total: 100
例:
def worker(tasks):
for task in tasks:
if task.lock.acquire(False):
time.sleep(0.01)
logging.info('{} {} begin to start'.format(threading.current_thread(), task.name))
else:
logging.info('{} {} 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, args=(tasks,), name='worker-{}'.format(i)).start()
输出:
……
2018-08-06-14:10:34 Thread info: 12256 worker-1 <Thread(worker-1, started 12256)> task-7 is working
2018-08-06-14:10:34 Thread info: 12256 worker-1 <Thread(worker-1, started 12256)> task-8 is working
2018-08-06-14:10:34 Thread info: 12256 worker-1 <Thread(worker-1, started 12256)> task-9 is working
2018-08-06-14:10:34 Thread info: 11912 worker-3 <Thread(worker-3, started 11912)> task-9 begin to start
2018-08-06-14:10:34 Thread info: 10496 worker-4 <Thread(worker-4, started 10496)> task-8 begin to start
2018-08-06-14:10:34 Thread info: 10496 worker-4 <Thread(worker-4, started 10496)> task-9 is working
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。