Python threading 多线程编程

import threading

class MyThread(threading.Thread):
    def run(self):
        print("线程开始执行")
        # 这里写线程要执行的代码
        for i in range(3):
            print(f"执行步骤 {i}")
            threading.Event().wait(0.5)  # 模拟工作
        print("线程执行结束")

# 创建并启动线程
thread = MyThread()
thread.start()
thread.join()
print("主线程结束")

import threading

def my_task():
    print("线程开始执行")
    for i in range(3):
        print(f"处理任务 {i}")
        threading.Event().wait(0.5)
    print("线程执行结束")

# 创建线程
thread = threading.Thread(target=my_task)
thread.start()
thread.join()
print("主线程结束")

import threading

lock = threading.Lock()
balance = 100

def update_balance(amount):
    global balance
    with lock:  # 自动获取和释放锁
        # 这部分代码同一时间只能有一个线程执行
        current = balance
        threading.Event().wait(0.01)  # 模拟处理时间
        balance = current + amount

# 创建多个线程同时修改余额
threads = []
for i in range(10):
    t = threading.Thread(target=update_balance, args=(10,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"最终余额: {balance}")  # 应该是 200

import threading
import queue
import time

def producer(q, name):
    for i in range(5):
        item = f"{name}生产的产品-{i}"
        q.put(item)
        print(f"生产者 {name}: {item}")
        time.sleep(0.1)

def consumer(q, name):
    while True:
        try:
            item = q.get(timeout=1)
            print(f"消费者 {name}: 处理 {item}")
            time.sleep(0.2)
            q.task_done()
        except queue.Empty:
            break

# 创建队列
q = queue.Queue()

# 创建生产者和消费者线程
producers = [
    threading.Thread(target=producer, args=(q, "A")),
    threading.Thread(target=producer, args=(q, "B"))
]

consumers = [
    threading.Thread(target=consumer, args=(q, "X")),
    threading.Thread(target=consumer, args=(q, "Y"))
]

# 启动所有线程
for p in producers:
    p.start()

for c in consumers:
    c.start()

# 等待生产者完成
for p in producers:
    p.join()

# 等待所有任务被处理
q.join()

print("所有任务完成")

import threading

# 创建事件
start_event = threading.Event()
finished_event = threading.Event()

def worker():
    print("工作线程等待开始信号...")
    start_event.wait()  # 等待事件
    print("工作线程开始工作")
    time.sleep(1)  # 模拟工作
    print("工作完成")
    finished_event.set()  # 发送完成信号

# 启动工作线程
threading.Thread(target=worker).start()

time.sleep(0.5)  # 主线程等待
print("主线程发送开始信号")
start_event.set()

print("主线程等待工作完成")
finished_event.wait()
print("所有工作完成")

import threading
import time

# 只允许3个线程同时访问
semaphore = threading.Semaphore(3)

def access_resource(thread_id):
    with semaphore:
        print(f"线程 {thread_id} 获得访问权限")
        time.sleep(1)  # 模拟资源使用
        print(f"线程 {thread_id} 释放访问权限")

# 创建多个线程尝试访问资源
threads = []
for i in range(8):
    t = threading.Thread(target=access_resource, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

import threading

def reminder():
    print("提醒：该休息一下了！")

# 5秒后执行提醒
timer = threading.Timer(5.0, reminder)
print("定时器已设置，5秒后提醒")
timer.start()

# 如果需要可以取消定时器
# timer.cancel()

import threading
import requests
import time

class Downloader:
    def __init__(self, max_workers=3):
        self.semaphore = threading.Semaphore(max_workers)
        self.results = {}
        self.lock = threading.Lock()
    
    def download_url(self, url, name):
        with self.semaphore:
            try:
                print(f"开始下载 {name}")
                response = requests.get(url, timeout=5)
                
                with self.lock:
                    self.results[name] = len(response.content)
                
                print(f"完成下载 {name}, 大小: {len(response.content)} 字节")
            except Exception as e:
                print(f"下载 {name} 失败: {e}")
    
    def start_downloads(self, url_list):
        threads = []
        for name, url in url_list:
            t = threading.Thread(target=self.download_url, args=(url, name))
            threads.append(t)
            t.start()
        
        for t in threads:
            t.join()
        
        return self.results

# 使用示例
if __name__ == "__main__":
    urls_to_download = [
        ("首页", "https://www.fly63.com"),
        ("文档", "https://www.fly63.com/docs"),
        ("示例", "https://www.fly63.com/examples")
    ]
    
    downloader = Downloader(max_workers=2)
    start_time = time.time()
    results = downloader.start_downloads(urls_to_download)
    end_time = time.time()
    
    print(f"下载完成，耗时: {end_time - start_time:.2f} 秒")
    print(f"结果: {results}")