生产者消费者的顺序
本节的东西:
Box(cap) Condition 保护的有界队列:put 满了等、get 空了等;等 = cond.wait_for(条件),等的时候把锁交出去、醒来再拿回;改完 notify_all
run_pc(cap, n) 一个生产者放 0..n-1、一个消费者取 n 个,交回消费者拿到的顺序容量 2 的队列、放 6 个、取 6 个,看消费者拿到的顺序,和跑完队列是不是空的、锁是不是放开的:
import threading
import time
class Counter:
def __init__(self):
self.n = 0
def inc(self):
tmp = self.n
time.sleep(0.001)
self.n = tmp + 1
class SafeCounter(Counter):
def __init__(self):
super().__init__()
self.lock = threading.Lock()
def inc(self):
with self.lock:
tmp = self.n
time.sleep(0.001)
self.n = tmp + 1
def hammer(counter, workers=2, times=30):
def job():
for _ in range(times):
counter.inc()
ts = [threading.Thread(target=job) for _ in range(workers)]
for t in ts:
t.start()
for t in ts:
t.join()
return counter.n
def probe(lock):
"""这把锁现在能不能立刻拿到?能就拿了再放回去,交回 True;拿不到交回 False。"""
if lock.acquire(blocking=False):
lock.release()
return True
return False
class Guard:
"""手写的 with 替身:进就 acquire,出就 release——不管是正常出还是异常出。"""
def __init__(self, lock):
self.lock = lock
def __enter__(self):
self.lock.acquire()
return self
def __exit__(self, exc_type, exc, tb):
self.lock.release()
return False
class Box:
"""有界队列:满了 put 等、空了 get 等;等的时候把锁交出去(wait_for 会放锁、醒来再拿回)。"""
def __init__(self, cap):
self.cap = cap
self.items = []
self.cond = threading.Condition()
def put(self, x):
with self.cond:
self.cond.wait_for(lambda: len(self.items) < self.cap)
self.items.append(x)
self.cond.notify_all()
def get(self):
with self.cond:
self.cond.wait_for(lambda: len(self.items) > 0)
x = self.items.pop(0)
self.cond.notify_all()
return x
def run_pc(cap, n):
"""一个生产者放 0..n-1,一个消费者取 n 个;交回消费者拿到的顺序。"""
box = Box(cap)
got = []
def producer():
for i in range(n):
box.put(i)
def consumer():
for _ in range(n):
got.append(box.get())
p = threading.Thread(target=producer, daemon=True)
c = threading.Thread(target=consumer, daemon=True)
p.start()
c.start()
p.join(timeout=2) # 写坏了的 get 会让生产者永远等——别把判题机拖死
c.join(timeout=2)
return got
class Gate:
"""最多 n 个同时在里面;顺便记下峰值(峰值计数器自己用一把锁保护)。"""
def __init__(self, n):
self.sem = threading.Semaphore(n)
self.lock = threading.Lock()
self.inside = 0
self.peak = 0
def __enter__(self):
self.sem.acquire()
with self.lock:
self.inside += 1
self.peak = max(self.peak, self.inside)
def __exit__(self, *a):
with self.lock:
self.inside -= 1
self.sem.release()
return False
def run_gate(n, workers, hold=0.03):
gate = Gate(n)
def job():
with gate:
time.sleep(hold)
ts = [threading.Thread(target=job) for _ in range(workers)]
for t in ts:
t.start()
for t in ts:
t.join()
return gate.peak
got = run_pc(2, 6)
box = Box(2)
box.put("a")
print("".join(str(x) for x in got) + "/" + str(box.get()) + "/" + str(len(box.items)) + "/" + str(probe(box.cond)))
全部评论