Python Thread类详解 - 自定义线程编程指南

# Thread类相关语法示例
import threading

# 方式1: 继承Thread类
class MyThread(threading.Thread):
    def __init__(self, name, args):
        super().__init__(name=name)
        self.args = args
    
    def run(self):
        # 线程执行逻辑
        print(f"线程 {self.name} 正在运行")
        # 使用 self.args 访问参数

# 方式2: 直接实例化Thread
def worker(name, delay):
    print(f"线程 {name} 开始")
    time.sleep(delay)
    print(f"线程 {name} 结束")

t = threading.Thread(target=worker, args=("Worker-1", 2))

# Thread类常用属性
t.name          # 线程名称
t.ident         # 线程ID
t.daemon        # 是否为守护线程
t.is_alive()    # 是否存活

# Thread类常用方法
t.start()       # 启动线程
t.join(timeout) # 等待线程结束
t.setName()     # 设置线程名
t.getName()     # 获取线程名

# 基本用法示例
import threading
import time

class DownloadThread(threading.Thread):
    """下载线程类"""
    def __init__(self, name, url, delay):
        super().__init__(name=name)
        self.url = url
        self.delay = delay
    
    def run(self):
        """线程执行方法"""
        print(f"[{self.name}] 开始下载: {self.url}")
        time.sleep(self.delay)  # 模拟下载时间
        print(f"[{self.name}] 下载完成: {self.url}")

# 创建多个下载线程
threads = []
urls = [
    ("https://example.com/file1.zip", 2),
    ("https://example.com/file2.zip", 3),
    ("https://example.com/file3.zip", 1),
]

for i, (url, delay) in enumerate(urls):
    t = DownloadThread(f"Download-{i+1}", url, delay)
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

print("所有下载任务完成")

# 详细代码示例：自定义Thread类实现生产者
import threading
import time
import random

class ProducerThread(threading.Thread):
    """生产者线程"""
    def __init__(self, name, queue, max_items):
        super().__init__(name=name)
        self.queue = queue
        self.max_items = max_items
    
    def run(self):
        """生产数据"""
        for i in range(self.max_items):
            item = f"{self.name}-产品-{i}"
            self.queue.append(item)
            print(f"[{self.name}] 生产: {item}")
            time.sleep(random.uniform(0.1, 0.5))
        
        print(f"[{self.name}] 生产完成")

class ConsumerThread(threading.Thread):
    """消费者线程"""
    def __init__(self, name, queue, consume_count):
        super().__init__(name=name)
        self.queue = queue
        self.consume_count = consume_count
    
    def run(self):
        """消费数据"""
        consumed = 0
        while consumed < self.consume_count:
            if self.queue:
                item = self.queue.pop(0)
                print(f"[{self.name}] 消费: {item}")
                consumed += 1
            time.sleep(random.uniform(0.2, 0.6))
        
        print(f"[{self.name}] 消费完成")

# 共享队列
shared_queue = []

# 创建生产者和消费者
producer = ProducerThread("Producer-1", shared_queue, 5)
consumer1 = ConsumerThread("Consumer-1", shared_queue, 3)
consumer2 = ConsumerThread("Consumer-2", shared_queue, 2)

# 启动线程
producer.start()
consumer1.start()
consumer2.start()

# 等待所有线程完成
producer.join()
consumer1.join()
consumer2.join()

print(f"最终队列状态: {shared_queue}")