Python 学习笔记(二)

1
2
3
4
5
6
7
8
import threadingdef worker(num):    print(f"Worker {num}")thread = threading.Thread(target=worker, args=(1,))thread.start()thread.join()

1
2
3
4
5
6
class MyThread(threading.Thread):    def run(self):        print("Thread running")thread = MyThread()thread.start()

1
2
3
# 设置守护线程thread = threading.Thread(target=func, daemon=True)thread.start()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import threadingcounter = 0def add():    global counter    for _ in range(100000):        counter += 1  # 非原子操作，存在竞态条件threads = [threading.Thread(target=add) for _ in range(2)]for t in threads:    t.start()for t in threads:    t.join()print(counter)  # 结果可能小于 200000

1
2
3
4
5
6
7
lock = threading.Lock()counter = 0def safe_add():    global counter    with lock:  # 自动 acquire() 和 release()        counter += 1

1
2
3
4
5
6
lock = threading.Lock()lock.acquire()try:    counter += 1  # 临界区代码finally:    lock.release()  # 确保锁释放

1
2
3
4
5
if lock.acquire(blocking=False):  # 不阻塞，立即返回    try:        pass    finally:        lock.release()

1
2
3
4
5
6
7
8
9
10
# 不好的写法：持有锁时间过长with lock:    do_something()    do_other_thing()    time.sleep(2)  # 不必在锁内# 更好的写法with lock:    update_shared_data()time.sleep(2)  # 锁外执行

1
2
3
4
5
6
lock = threading.RLock()def recursive_function():    with lock:        print("进入临界区")        recursive_function()  # 同一线程可以再次获取锁

1
2
3
4
5
semaphore = threading.Semaphore(2)  # 允许最多2个线程同时访问def access_resource():    with semaphore:        print(f"Thread {threading.current_thread().name} accessing resource")

1
2
3
4
5
6
7
8
9
10
11
12
13
condition = threading.Condition()queue = []def producer():    with condition:        queue.append("item")        condition.notify()  # 通知等待的线程def consumer():    with condition:        while not queue:            condition.wait()  # 等待队列非空        item = queue.pop(0)

1
2
3
4
5
6
7
8
9
event = threading.Event()def worker():    print("Waiting for event...")    event.wait()  # 等待事件被设置    print("Event fired!")# 在主线程中触发事件event.set()

1
2
3
4
5
6
7
8
9
import queueq = queue.Queue(maxsize=10)def producer():    q.put("item")def consumer():    item = q.get()  # 自动处理锁与条件变量

1
2
3
4
5
6
7
8
9
10
from dataclasses import dataclass@dataclassclass Task:    id: str    subject: str    description: str    status: str          # pending | in_progress | completed    owner: str | None    # Agent name    blockedBy: list[str]  # Dependency task IDs

1
2
3
4
5
6
7
8
9
@dataclassclass Task:    id: str    status: str = "pending"    def __post_init__(self):        # 初始化后自动验证        if self.status not in ["pending", "in_progress", "completed"]:            raise ValueError(f"Invalid status: {self.status}")

1
2
3
4
5
import jsonfrom pathlib import Pathdef load_task(task_id: str) -> Task:    return Task(**json.loads(_task_path(task_id).read_text()))

1
2
data = {"name": "Alice", "age": 30}# Person(**data) 等价于 Person(name="Alice", age=30)