忽然发现同事老婆很漂亮，感觉身边研发同事好多这样，人都不善言辞的，找老婆倒是都很成功。

刚看到个贴子，说是忽然发现同事老婆巨漂亮，回头一看，身边一堆研发同事都是这种：人闷声不爱说话，找对象一个比一个成功，网友都在感叹“原来闷骚男是婚恋大热门”。

我觉得这事吧，表面看是“程序员都娶漂亮老婆”，本质还是“稳定感”和“稀缺资源”的问题。研发普遍收入还行、人也专一，作息规律，情绪波动小，在婚恋市场上就很吃香。

网友们有的调侃“果然话少的钱多更受欢迎”，也有人羡慕自己怎么没这种属性。我比较认同一点：别把“找个漂亮老婆”当成炫耀资本。婚姻又不是集邮，好看是加分项，真正能走远的，还是性格、责任感、共同的生活能力

算法题：设计有限阻塞队列

先说个我前两天刚遇到的事儿。

那天晚上快十一点，我在公司楼下等外卖，手机远程连着一台测试机跑压测，结果一看日志，全是“队列满了”“等待可用空间”这种提示。几个生产者线程往里疯狂塞数据，消费者那边又跟不上节奏。那一刻我才想起来：啧，还是得老老实实搞个“有限阻塞队列”，不然线程之间靠 if / sleep 协调，迟早把自己绕晕。

你可以先脑补这么个画面：一条生产线，最多只能放 5 个箱子。工人 A 负责往上传箱子，工人 B 负责把箱子往下搬。台面上有 0 个箱子的时候，A 不能干活了吗？不是，是 B 没活干；反过来台面上已经有 5 个箱子了，A 就得停下等等 B 搬走一个。这个“最多 5 个”“空了要等”“满了也要等”的东西，其实就是我们要实现的有限阻塞队列。

落到代码里，核心就三样东西：一个真正存数据的容器，一个锁保证线程别打架，再加上两个“条件变量”让线程可以老老实实睡觉、被别人叫醒，而不是傻傻 while True + sleep。Python 里直接用 threading.Lock 和 threading.Condition 就够了。

我先把完整代码丢出来，你大概扫一眼，后面再一点点拆开讲：

import threading
import time
from collections import deque


classBoundedBlockingQueue:
def__init__(self, capacity: int):
if capacity <= 0:
raise ValueError("capacity must be positive")
        self.capacity = capacity
        self.queue = deque()

# 一把锁，两个条件：不满 / 不空
        self.lock = threading.Lock()
        self.not_full = threading.Condition(self.lock)
        self.not_empty = threading.Condition(self.lock)

defput(self, item):
"""放数据，队列满了就阻塞等待"""
with self.not_full:  # 进入时会先获取 self.lock
# 用 while 而不是 if，防止虚假唤醒
while len(self.queue) >= self.capacity:
                self.not_full.wait()
# 这里一定是有空间了
            self.queue.append(item)
# 放进去之后，至少有一个消费者可以被叫醒了
            self.not_empty.notify()

defget(self):
"""取数据，队列空就阻塞等待"""
with self.not_empty:
while len(self.queue) == 0:
                self.not_empty.wait()
            item = self.queue.popleft()
# 取走一个之后，可能有生产者在等空间
            self.not_full.notify()
return item

defsize(self) -> int:
"""当前队列里元素个数"""
with self.lock:
return len(self.queue)


# 简单测一把：多个生产者 + 多个消费者
defproducer(q: BoundedBlockingQueue, pid: int, n: int):
for i in range(n):
        item = f"p{pid}-{i}"
        q.put(item)
        print(f"[producer-{pid}] put {item}, size={q.size()}")
        time.sleep(0.1)  # 模拟生产耗时


defconsumer(q: BoundedBlockingQueue, cid: int):
whileTrue:
        item = q.get()
        print(f"    [consumer-{cid}] got {item}, size={q.size()}")
        time.sleep(0.3)  # 模拟消费耗时


if __name__ == "__main__":
    q = BoundedBlockingQueue(capacity=5)

# 起两个消费者
for cid in range(2):
        t = threading.Thread(target=consumer, args=(q, cid), daemon=True)
        t.start()

# 起三个生产者，每个生产 5 个
    producers = []
for pid in range(3):
        t = threading.Thread(target=producer, args=(q, pid, 5))
        t.start()
        producers.append(t)

# 等生产者生产完
for t in producers:
        t.join()

    print("all producers finished, wait a bit...")
    time.sleep(2)

你可以自己在本地跑一下，大概会看到这种效果：队列 size 在 0～5 之间来回波动，生产太快的时候会卡在 put 上等消费者，消费太快的时候会卡在 get 上等生产者，这就说明“阻塞”这事儿生效了。

这里有几个地方，稍微注意下，不然很容易写着写着就出 bug：

第一个就是那个 while。很多人第一次写会习惯性写成 if len(self.queue) == 0: self.not_empty.wait()，看起来也没问题，但条件变量有个“虚假唤醒”的坑：线程从 wait() 返回不一定真的是有人叫醒，有可能是系统自己莫名其妙把你唤醒了。所以标准写法都是 while 条件不满足: wait()，醒来以后再检查一遍条件，条件没满足就继续睡。

第二个是锁和条件用同一把。上面代码里 Condition(self.lock) 就是这个意思：进入 with self.not_empty 的时候，会先拿到锁，wait() 的时候会暂时把锁放掉，让别的线程进来修改队列；等被唤醒的时候又会把锁重新拿回来，然后再往下走。要是你锁和条件乱配，线程之间就会出现各种奇怪的“互相等着对方释放锁”的僵局。

第三个是容量的语义。现在我们是“硬容量”，一旦达到 capacity 就必须等。你也可以很容易加个不阻塞版本，比如：

deftry_put(self, item) -> bool:
with self.not_full:
if len(self.queue) >= self.capacity:
returnFalse
            self.queue.append(item)
            self.not_empty.notify()
returnTrue

这样在一些“宁可丢也不能卡”的场景，比如打日志、上报监控，你就可以先 try_put，失败了就直接丢掉，主流程继续跑，别被队列拖慢。

再往上一点想，其实 Python 标准库里已经给你准备好了一个一模一样的东西：queue.Queue(maxsize=N)，里面就是类似的锁 + 条件变量实现。自己手写这个有限阻塞队列，更像是一道“数据结构 + 并发”的综合题，顺便把生产者-消费者问题吃透一遍。

行，差不多就这样，我先去冲杯咖啡，你把上面这个类改改、比如加个超时参数、加个关闭标志啥的，基本这个算法你就算真掌握了。