Python多线程与多进程详解：性能提升技巧与实战案例

import requests
from concurrent.futures import ThreadPoolExecutor

def download_url(url):
    response = requests.get(url)
    return response.content

urls = ['http://example.com', 'http://example.org', 'http://example.net']

with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(download_url, urls))

print("下载完成")

from concurrent.futures import ProcessPoolExecutor

def square_number(n):
    return n * n

numbers = list(range(1000000))

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(square_number, numbers))

print("计算完成")

import requests
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def download_url(url):
    response = requests.get(url)
    return response.content

def process_data(data):
    # 假设这是一个CPU密集型的处理任务
    return len(data)

urls = ['http://example.com', 'http://example.org', 'http://example.net']

# 使用多线程下载数据
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(download_url, urls))

# 使用多进程处理数据
with ProcessPoolExecutor(max_workers=4) as executor:
    processed_results = list(executor.map(process_data, results))

print("下载和处理完成")

import requests
from bs4 import BeautifulSoup
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def download_url(url):
    response = requests.get(url)
    return response.text

def extract_text(html):
    soup = BeautifulSoup(html, 'html.parser')
    return soup.get_text()

def count_words(text):
    return len(text.split())

urls = ['http://example.com', 'http://example.org', 'http://example.net']

# 使用多线程下载网页
with ThreadPoolExecutor(max_workers=3) as executor:
    html_contents = list(executor.map(download_url, urls))

# 使用多进程提取文本并统计单词数量
with ProcessPoolExecutor(max_workers=4) as executor:
    texts = list(executor.map(extract_text, html_contents))
    word_counts = list(executor.map(count_words, texts))

print("网页下载和数据处理完成")
print("单词统计：", word_counts)

from multiprocessing import Process, Queue

def producer(queue):
    for i in range(10):
        queue.put(i)
    queue.put(None)  # 结束信号

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

queue = Queue()
producer_process = Process(target=producer, args=(queue,))
consumer_process = Process(target=consumer, args=(queue,))

producer_process.start()
consumer_process.start()

producer_process.join()
consumer_process.join()

from concurrent.futures import ThreadPoolExecutor, as_completed

def process_data(data):
    return sum(data)

datasets = [range(1000), range(2000), range(3000)]

with ThreadPoolExecutor(max_workers=3) as executor:
    future_to_data = {executor.submit(process_data, data): data for data in datasets}
    for future in as_completed(future_to_data):
        result = future.result()
        print(f"处理结果: {result}")

import asyncio
from concurrent.futures import ProcessPoolExecutor

async def fetch_url(url):
    await asyncio.sleep(1)  # 模拟IO操作
    return f"Fetched data from {url}"

def process_data(data):
    return len(data)

async def main(urls):
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        # 异步获取数据
        tasks = [fetch_url(url) for url in urls]
        fetched_data = await asyncio.gather(*tasks)
        
        # 使用多进程处理数据
        processed_data = await loop.run_in_executor(pool, lambda: [process_data(data) for data in fetched_data])
        
        print("处理完成")
        print("处理结果:", processed_data)

urls = ['http://example.com', 'http://example.org', 'http://example.net']
asyncio.run(main(urls))

from multiprocessing import Process, Lock

def task(lock):
    with lock:
        print("任务开始")
        # 模拟任务
        import time
        time.sleep(1)
        print("任务结束")

lock = Lock()
processes = [Process(target=task, args=(lock,)) for _ in range(4)]

for p in processes:
    p.start()
for p in processes:
    p.join()

import os
from multiprocessing import Pool

def compute_task(x):
    return x * x

if __name__ == "__main__":
    cpu_count = os.cpu_count()
    with Pool(processes=cpu_count) as pool:
        results = pool.map(compute_task, range(1000))
    print("结果:", results)

from threading import Thread, Lock

counter = 0
lock = Lock()

def increment():
    global counter
    for _ in range(1000000):
        with lock:
            counter += 1

threads = [Thread(target=increment) for _ in range(4)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

print(f"最终计数值: {counter}")

from queue import Queue
from threading import Thread

def producer(queue):
    for i in range(10):
        queue.put(i)
    queue.put(None)  # 结束信号

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

queue = Queue()
producer_thread = Thread(target=producer, args=(queue,))
consumer_thread = Thread(target=consumer, args=(queue,))

producer_thread.start()
consumer_thread.start()

producer_thread.join()
consumer_thread.join()