线程 VS 协程

import logging
import threading
import time

# 配置日志格式
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s',
    datefmt='%H:%M:%S'
)

def worker(event: threading.Event):
    logging.info("子线程等待异步事件...")
    logging.info(f"event is set? {event.is_set()}")
    event.wait()  # 触发事件
    logging.info(f"event is set? {event.is_set()}")
    logging.info("子线程收到事件信号、处理任务并退出")

if __name__ == "__main__":
    event = threading.Event()
    t = threading.Thread(target=worker, args=(event,), daemon=True)
    t.start()

    # 主线程延迟后发送信号
    logging.info("主线程开始...")
    
    time.sleep(2)
    event.set()  # 等待事件触发
    t.join()
    logging.info("主线程结束...")

import threading
import logging
import time

# 配置日志格式
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s',
    datefmt='%H:%M:%S'
)

# 全局计数器
counter = 0


def worker(lock: threading.Lock, name: str):
    global counter
    with lock:  # 自动调用acquire()，退出时自动调用release()
        counter += 1
        logging.info(f"Worker {name} 获得锁，开始操作，计数器值: {counter}")
        time.sleep(1)  # 模拟耗时操作
        logging.info(f"Worker {name} 释放锁")


def main():
    global counter
    lock = threading.Lock()
    threads = []
    
    # 创建10个线程
    for i in range(10):
        thread = threading.Thread(target=worker, args=(lock, f"{i+1}"))
        threads.append(thread)
        thread.start()
    
    # 等待所有线程完成
    for thread in threads:
        thread.join()
    
    logging.info(f"所有任务完成，最终计数器值: {counter}")


if __name__ == "__main__":
    main()

import threading
import time
import logging

# 配置日志格式
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s',
    datefmt='%H:%M:%S'
)

class SharedBuffer:
    def __init__(self):
        self.buffer = []
        self.cond = threading.Condition()  # 条件变量
        self.max_size = 3

    def produce(self, item):
        with self.cond:
            # 缓冲区满时等待
            while len(self.buffer) == self.max_size:
                self.cond.wait()
            self.buffer.append(item)
            logging.info(f"生产: {item}, 缓冲区: {self.buffer}")
            self.cond.notify()  # 通知消费者

    def consume(self):
        with self.cond:
            # 缓冲区空时等待
            while not self.buffer:
                self.cond.wait()
            item = self.buffer.pop(0)
            logging.info(f"消费: {item}, 缓冲区: {self.buffer}")
            self.cond.notify()  # 通知生产者
            return item

def producer(buffer):
    for i in range(10):
        buffer.produce(i)
        time.sleep(1)

def consumer(buffer):
    for _ in range(10):
        buffer.consume()
        time.sleep(2)

if __name__ == "__main__":
    buffer = SharedBuffer()
    t1 = threading.Thread(target=producer, args=(buffer,))
    t2 = threading.Thread(target=consumer, args=(buffer,))
    t1.start()
    t2.start()
    t1.join()
    t2.join()

import threading
import logging
import time

# 配置日志格式
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s',
    datefmt='%H:%M:%S'
)


# 定义一个需要控制并发的函数
def limited_task(sem, task_id):
    # 通过with自动获取和释放许可
    with sem:
        logging.info(f"任务 {task_id} 开始执行")
        time.sleep(2)  # 模拟耗时操作（如网络请求）
        logging.info(f"任务 {task_id} 执行完成")


def main():
    # 初始化信号量，最多允许3个并发线程
    sem = threading.Semaphore(3)
    
    # 创建10个任务（线程）
    threads = []
    for i in range(10):
        thread = threading.Thread(target=limited_task, args=(sem, i))
        threads.append(thread)
        thread.start()
    
    # 等待所有线程完成
    for thread in threads:
        thread.join()


if __name__ == "__main__":
    main()

import threading
import time
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s',
    datefmt='%H:%M:%S'
)

def worker(barrier, name, phase1_duration, phase2_duration):
    # 第一阶段
    logging.info(f"{name} 开始第一阶段")
    time.sleep(phase1_duration)  # 模拟工作
    barrier.wait()  # 等待所有worker到达第一阶段
    logging.info(f"{name} 结束第一阶段")
    
    # 第二阶段
    logging.info(f"{name} 开始第二阶段")
    time.sleep(phase2_duration)  # 模拟工作
    barrier.wait()  # 等待所有worker到达第二阶段
    logging.info(f"{name} 结束第二阶段")

def main():
    num_workers = 3
    barrier = threading.Barrier(num_workers)  # 3个参与者
    
    # 创建线程
    threads = []
    workers = [
        ("Worker 1", 1, 2),
        ("Worker 2", 2, 3),
        ("Worker 3", 3, 4)
    ]
    
    for name, phase1_duration, phase2_duration in workers:
        thread = threading.Thread(
            target=worker,
            args=(barrier, name, phase1_duration, phase2_duration)
        )
        threads.append(thread)
        thread.start()
    
    # 等待所有线程完成
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    main()