Python3 多任务并发编程的实现

import threading
import time

def worker(name, delay):
    for i in range(3):
        print(f"{name} 第 {i+1} 次")
        time.sleep(delay)

t1 = threading.Thread(target=worker, args=("A", 1))
t2 = threading.Thread(target=worker, args=("B", 2))

t1.start()
t2.start()

t1.join()  # 等 t1 结束
t2.join()  # 等 t2 结束
print("全部完成")

class DownloadTask(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = None
    
    def run(self):
        # 线程启动后自动执行
        print(f"下载：{self.url}")
        time.sleep(2)
        self.result = f"{self.url} 完成"

task = DownloadTask("https://example.com/file.zip")
task.start()
task.join()
print(task.result)

def background():
    while True:
        print("后台运行...")
        time.sleep(1)

t = threading.Thread(target=background, daemon=True)
t.start()

time.sleep(3)
print("主程序结束")  # 3秒后退出，守护线程自动终止

counter = 0

def increment():
    global counter
    for _ in range(100000):
        counter += 1

t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
t1.start()
t2.start()
t1.join()
t2.join()

print(counter)  # 期望 200000，实际可能 189xxx

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:   # 拿到锁再操作
            counter += 1
        # 离开 with 自动释放锁

import queue
import threading

q = queue.Queue()

def producer():
    for i in range(5):
        q.put(i)
        print(f"生产：{i}")
    q.put(None)  # 结束信号

def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消费：{item}")

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start()
t2.start()
t1.join()
t2.join()

from concurrent.futures import ThreadPoolExecutor

def task(n):
    return n * 2

# 创建线程池，最多3个线程同时跑
with ThreadPoolExecutor(max_workers=3) as executor:
    # 方式1：提交单个任务
    future = executor.submit(task, 5)
    print(future.result())  # 10
    
    # 方式2：批量提交
    results = executor.map(task, range(5))
    print(list(results))  # [0, 2, 4, 6, 8]

# 获取完成的任务（谁先完成先处理谁）
with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(5)]
    for future in as_completed(futures):
        print(future.result())

from concurrent.futures import ThreadPoolExecutor
import requests
import time

urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/2",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/3",
]

def download(url):
    start = time.time()
    resp = requests.get(url, timeout=10)
    elapsed = time.time() - start
    return {"url": url, "status": resp.status_code, "time": elapsed}

# 串行
start = time.time()
for url in urls:
    print(download(url))
print(f"串行耗时：{time.time() - start:.2f}s")

# 并行（线程池）
start = time.time()
with ThreadPoolExecutor(max_workers=4) as executor:
    results = executor.map(download, urls)
    for r in results:
        print(r)
print(f"并行耗时：{time.time() - start:.2f}s")