Python技术迷

大厂员工:公司花月薪6万,挖来一位技术大牛,结果入职还没满两个月就和部门负责人正面杠上了,导火索只有一个,他不怎么加班!

刚看到个贴子,说大厂花月薪6万挖来个技术大牛,结果不到两个月就和部门负责人吵起来,就因为——他不怎么加班。

Image

我觉得这事吧,关键在预期没对齐。公司心里想的是:高薪=听话+卖时间;大牛心里想的是:高薪=买结果+尊重边界。想的都不一样,迟早要爆。

从我的角度看,高薪确实意味着责任更重,但责任应该体现在结果上,而不是工位上坐多久。公司要的是产出,就该用目标和机制说话;别总靠“奋斗文化”道德绑架。

不过话说回来,大牛入职前也该打听清楚团队节奏,真是全年无休那种,就别指望自己一个人能改变文化了。

算法题:交替打印 FooBar

说个很日常的场景哈:你在公司写脚本,老板说“一个线程负责打日志里的 Foo,一个线程负责打 Bar,输出结果必须是 FooBarFooBar... 不能乱”,嘴上说是小需求,实际上就是在考你多线程同步那一套。这个在刷题网站上就叫:交替打印 FooBar。

题目先说清楚

大致意思就是:

  • 有两个线程,一个只负责打印 "Foo"
  • 另一个只负责打印 "Bar"
  • 一共打印 n 次
  • 最终输出必须是:FooBarFooBar... 这样的顺序

伪代码一般长这样(Python 版):

classFooBar:
def__init__(self, n: int):
        self.n = n

deffoo(self, printFoo):# printFoo 是个函数
# 循环 n 次打印 "Foo"
pass

defbar(self, printBar):# printBar 是个函数
# 循环 n 次打印 "Bar"
pass

真正难点不在打印,而在“两个线程交替”这四个字:不管系统怎么调度线程,你都得保证 Foo 一定在 Bar 前面,不能连着两个 Foo 或连着两个 Bar。


核心思路:谁先谁后,用“同步原语”规定死

Python 里多线程要控顺序,常见的工具就几种:Lock / RLock、Condition、Semaphore、Event 等等。

这个题本质是:

两个工人抢着干活,但必须轮流,你就得想办法设计一个“红绿灯”。

下面给两个比较常用、也好理解的写法。


写法一:Condition + 标志位

这个思路很像排队叫号机:

  • 用一个布尔变量 self.foo_turn 表示“轮到谁”
  • 两个线程共用一把锁 + 一个条件变量 Condition
  • 不该自己打印的时候就 wait(),轮到自己时打印完再 notify() 叫醒对方

代码来一版,直接可跑:

import threading


classFooBar:
def__init__(self, n: int):
        self.n = n
        self.lock = threading.Lock()
        self.cond = threading.Condition(self.lock)
        self.foo_turn = True# 一开始先打印 Foo

deffoo(self, printFoo):
for _ in range(self.n):
with self.cond:  # 自动加锁/解锁
# 不是我该打印就等着
whilenot self.foo_turn:
                    self.cond.wait()
# 轮到我了
                printFoo()
# 轮到 Bar
                self.foo_turn = False
# 通知对方可以干活
                self.cond.notify()

defbar(self, printBar):
for _ in range(self.n):
with self.cond:
while self.foo_turn:  # 还没轮到我
                    self.cond.wait()
                printBar()
# 下一轮又轮到 Foo
                self.foo_turn = True
                self.cond.notify()

简单测一下:

defprintFoo():
    print("Foo", end='')

defprintBar():
    print("Bar", end='')


if __name__ == "__main__":
    fb = FooBar(5)
    t1 = threading.Thread(target=fb.foo, args=(printFoo,))
    t2 = threading.Thread(target=fb.bar, args=(printBar,))

    t1.start()
    t2.start()
    t1.join()
    t2.join()
# 输出: FooBarFooBarFooBarFooBarFooBar

这个版本有几个点可以注意一下:

  • while 不是 if:wait() 被唤醒不代表一定轮到你了,可能是“虚惊一场”,所以醒来要再检查一次条件
  • Condition 必须配合锁用: 进入 with self.cond 时会先拿锁,wait() 内部会释放锁并睡眠,被唤醒后再重新拿锁继续执行
  • 标志位一定要在锁保护下修改,不然两个线程一起改就乱了

写法二:两个 Semaphore,像接力棒

还有一个更“干脆”的办法,用 Semaphore(信号量),它内部其实就是一个带计数的锁,非常适合做交替这种场景。

思路是这样的:

  • 准备两个信号量:

    • foo_sem 初始值为 1,表示一开始 Foo 可以先跑
    • bar_sem 初始值为 0,表示 Bar 先别急
  • Foo 每次:

  1. foo_sem.acquire() 拿到“令牌”才打印
  2. 打印 "Foo"
  3. bar_sem.release() 把令牌交给 Bar
  • Bar 每次:

    1. bar_sem.acquire() 等着 Foo 把令牌给它
    2. 打印 "Bar"
    3. foo_sem.release() 把令牌交回 Foo

    代码也给你写好:

    import threading


    classFooBar:
    def__init__(self, n: int):
            self.n = n
            self.foo_sem = threading.Semaphore(1)  # 先让 Foo 走
            self.bar_sem = threading.Semaphore(0)  # Bar 先阻塞

    deffoo(self, printFoo):
    for _ in range(self.n):
                self.foo_sem.acquire()  # 等自己“绿灯”
                printFoo()
                self.bar_sem.release()  # 放行 Bar

    defbar(self, printBar):
    for _ in range(self.n):
                self.bar_sem.acquire()  # 等 Foo 放行
                printBar()
                self.foo_sem.release()  # 放行下一轮 Foo

    同样跑一下就能看到 FooBarFooBar...。

    这个写法比 Condition 那个还直观一点: 你就把两个信号量当成两把“接力棒”,谁拿到棒谁跑,跑完把棒丢给对方。

    -END-

    我为大家打造了一份RPA教程,完全免费:songshuhezi.com/rpa.html

    🔥虎哥私藏精品🔥

    虎哥作为一名老码农,整理了全网最全《python高级架构师资料合集》,总量高达650GB