基于Python的性能优化（线程、协程、进程）

import threading
import time
import requests

def fetch_url(url: str)->None:
    '''根据地址发起请求，获取响应
    - url: 请求地址'''
    response = requests.get(url)
    print(f"{url}: {response.status_code}")

def fetch_urls_sequential(urls:list)->None:
    start_time = time.time()
    for url in urls:
        fetch_url(url)
    end_time = time.time()
    print(f"使用单线程时间为: {end_time - start_time} 秒\n")

def fetch_urls_concurrent(urls:list)->None:
    start_time = time.time()
    threads = []
    for url in urls:
        thread = threading.Thread(target=fetch_url, args=(url,))
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()
    end_time = time.time()
    print(f"使用多线程时间为: {end_time - start_time} 秒")

if __name__ == "__main__":
    urls = ["http://www.example.com"]*10
    fetch_urls_sequential(urls)
    fetch_urls_concurrent(urls)

import asyncio
import threading
import aiohttp
import time
import requests

async def fetch_page(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()
        
async def compute_length(url):
    page = await fetch_page(url)
    return len(page)

async def async_main():
    urls = ["http://www.example.com"]*10
    tasks = [compute_length(url) for url in urls]
    results = await asyncio.gather(*tasks)
    print("内容长度为:", results)

def thread_fetch_page(url):
    response = requests.get(url)
    return response.text
def thread_compute_length(url):
    page = thread_fetch_page(url)
    return len(page)
def thread_main():
    urls = ["http://www.example.com"]*10    
    threads = []
    results = []
    start_time = time.time()
    for url in urls:
        thread = threading.Thread(target=lambda u: results.append(thread_compute_length(u)), args=(url,))
        threads.append(thread)
        thread.start()
    for thread in threads:
        thread.join()
    end_time = time.time()
    print("内容长度为:", results)
    print(f"使用多线程时间为: {end_time - start_time} 秒")    

if __name__ == "__main__":
    start_time = time.time()
    asyncio.run(async_main())
    end_time = time.time()
    print(f"使用协程时间为: {end_time - start_time} 秒")
    thread_main()

import multiprocessing
import threading
import numpy as np
import time

def integrate(f, a, b, N):
    """使用梯形法则计算f在区间[a, b]上的积分,N为分割数"""
    x = np.linspace(a, b, N)
    y = f(x)
    dx = (b - a) / (N - 1)
    return np.trapz(y, dx=dx)

def f(x):
    '''计算积分'''
    return np.sin(x) * np.exp(-x)

def integrate_range(start, end, result, index):
    result[index] = integrate(f, start, end, 100000000) 

def thread_main():
    result = [None] * 4
    threads = []
    ranges = [(0, 5), (5, 10), (10, 15), (15, 20)]  

    start_time = time.time()

    for i, (start, end) in enumerate(ranges):
        thread = threading.Thread(target=integrate_range, args=(start, end, result, i))
        threads.append(thread)
        thread.start()

    for thread in threads:
        thread.join()

    end_time = time.time()
    print(f"多线程使用时间为: {end_time - start_time} 秒")
    print(f"积分结果: {result}\n")
def multiprocess_main():
    manager = multiprocessing.Manager()
    result = manager.list([None] * 4)
    processes = []
    ranges = [(0, 5), (5, 10), (10, 15), (15, 20)]  
    start_time = time.time()

    for i, (start, end) in enumerate(ranges):
        process = multiprocessing.Process(target=integrate_range, args=(start, end, result, i))
        processes.append(process)
        process.start()

    for process in processes:
        process.join()

    end_time = time.time()
    print(f"多进程使用时间为: {end_time - start_time} 秒")
    print(f"积分结果: {result}")

if __name__ == "__main__":
    thread_main()    
    multiprocess_main()