Python 并发编程的高级技巧与性能优化(最佳实践)

from concurrent.futures import ThreadPoolExecutor
import time
def task(n):
    time.sleep(1)
    return n * n
def main():
    # 创建线程池
    with ThreadPoolExecutor(max_workers=4) as executor:
        # 提交任务
        futures = [executor.submit(task, i) for i in range(10)]
        # 获取结果
        results = [future.result() for future in futures]
    print(f"Results: {results}")
if __name__ == "__main__":
    start = time.time()
    main()
    end = time.time()
    print(f"Time taken: {end - start:.2f} seconds")

import asyncio
import aiohttp
async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()
async def main():
    urls = [
        "https://api.github.com/users/octocat",
        "https://api.github.com/users/github",
        "https://api.github.com/users/pytorch",
        "https://api.github.com/users/tensorflow"
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    print(f"Fetched {len(results)} URLs")
if __name__ == "__main__":
    import time
    start = time.time()
    asyncio.run(main())
    end = time.time()
    print(f"Time taken: {end - start:.2f} seconds")

from concurrent.futures import ProcessPoolExecutor
import time
def cpu_bound_task(n):
    result = 0
    for i in range(n):
        result += i * i
    return result
def main():
    # 创建进程池
    with ProcessPoolExecutor(max_workers=4) as executor:
        # 提交任务
        futures = [executor.submit(cpu_bound_task, 10**7) for _ in range(4)]
        # 获取结果
        results = [future.result() for future in futures]
    print(f"Results: {results}")
if __name__ == "__main__":
    start = time.time()
    main()
    end = time.time()
    print(f"Time taken: {end - start:.2f} seconds")

import threading
import time
# 共享资源
counter = 0
lock = threading.Lock()
def increment():
    global counter
    for _ in range(1000000):
        with lock:
            counter += 1
def main():
    # 创建线程
    threads = [threading.Thread(target=increment) for _ in range(4)]
    # 启动线程
    for t in threads:
        t.start()
    # 等待线程完成
    for t in threads:
        t.join()
    print(f"Counter: {counter}")
if __name__ == "__main__":
    start = time.time()
    main()
    end = time.time()
    print(f"Time taken: {end - start:.2f} seconds")