北京百思可瑞教育：Python进程与多进程编程，Process语法、PID/PPID用法及concurrent.futures模块详解

pythonimport multiprocessingimport time def worker(task_id, duration):    print(f"进程 {multiprocessing.current_process().pid} 执行任务 {task_id}")    time.sleep(duration)    return f"任务{task_id}完成" if __name__ == '__main__':    processes = []    for i in range(3):        p = multiprocessing.Process(            target=worker,            args=(i, 2),  # 位置参数必须为元组            name=f"Worker-{i}"        )        processes.append(p)        p.start()        for p in processes:        p.join()  # 阻塞主进程直到所有子进程结束

pythonclass CustomProcess(multiprocessing.Process):    def __init__(self, task_id):        super().__init__()        self.task_id = task_id        def run(self):        print(f"自定义进程 {self.pid} 处理任务 {self.task_id}") if __name__ == '__main__':    p1 = CustomProcess(101)    p2 = CustomProcess(102)    p1.start()    p2.start()    p1.join()    p2.join()

pythondef background_task():    while True:        print("后台服务运行中...")        time.sleep(1) if __name__ == '__main__':    p = multiprocessing.Process(target=background_task)    p.daemon = True  # 必须设置在start()之前    p.start()    time.sleep(3)    print("主进程退出，守护进程自动终止")

pythonimport psutil def print_process_tree(pid):    try:        p = psutil.Process(pid)        print(f"PID: {p.pid}, PPID: {p.ppid()}, Name: {p.name()}")        for child in p.children(recursive=True):            print(f"  ├─ Child PID: {child.pid}")    except psutil.NoSuchProcess:        print(f"进程 {pid} 不存在") if __name__ == '__main__':    print_process_tree(os.getpid())  # 打印当前进程树

pythondef monitor_process(pid, interval=1):    p = psutil.Process(pid)    while True:        try:            mem = p.memory_info().rss / 1024**2  # MB            cpu = p.cpu_percent(interval=interval)            print(f"PID: {pid}, CPU: {cpu:.1f}%, MEM: {mem:.2f}MB")        except psutil.NoSuchProcess:            break

pythonimport mathfrom concurrent.futures import ProcessPoolExecutor def is_prime(n):    if n < 2:        return False    for i in range(2, int(math.sqrt(n)) + 1):        if n % i == 0:            return False    return True def find_primes(start, end):    with ProcessPoolExecutor(max_workers=4) as executor:        numbers = range(start, end + 1)        # 方法1：使用submit获取Future对象        futures = [executor.submit(is_prime, num) for num in numbers]        for future in futures:            if future.result():                print(f"发现素数: {future.result()}")  # 实际应获取输入参数                # 方法2：使用map直接获取结果（更简洁）        results = executor.map(is_prime, numbers)        primes = [num for num, res in zip(numbers, results) if res]        print(f"{start}-{end}范围内的素数: {primes}") if __name__ == '__main__':    find_primes(100, 200)

pythondef risky_operation(x):    if x == 3:        raise ValueError("故意抛出异常")    return x * 2 with ProcessPoolExecutor() as executor:    try:        future = executor.submit(risky_operation, 3)        result = future.result(timeout=1)  # 设置超时    except ValueError as e:        print(f"捕获到异常: {e}")    except concurrent.futures.TimeoutError:        print("操作超时")

pythondef producer(queue):    for i in range(5):        queue.put(f"数据-{i}")        print(f"生产数据-{i}") def consumer(queue):    while True:        item = queue.get()        if item is None:  # 终止信号            break        print(f"消费 {item}") if __name__ == '__main__':    q = multiprocessing.Queue()    p = multiprocessing.Process(target=producer, args=(q,))    c = multiprocessing.Process(target=consumer, args=(q,))    p.start()    c.start()    p.join()    q.put(None)  # 发送终止信号    c.join()

pythonlock = multiprocessing.Lock()shared_counter = multiprocessing.Value('i', 0) def increment():    with lock:  # 自动获取和释放锁        for _ in range(10000):            shared_counter.value += 1 if __name__ == '__main__':    processes = [multiprocessing.Process(target=increment) for _ in range(4)]    for p in processes:        p.start()    for p in processes:        p.join()    print(f"最终计数: {shared_counter.value}")  # 应为40000

pythonimport osfrom concurrent.futures import ProcessPoolExecutor def process_data(data):    return sum(data) if __name__ == '__main__':    data_chunks = [[i, i+1, i+2] for i in range(0, 100, 3)]    with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:        results = list(executor.map(process_data, data_chunks))    print(f"总结果: {sum(results)}")