Python并發編程性能測試

在Python中進行并發編程性能測試時，可以使用concurrent.futures模塊中的ThreadPoolExecutor和ProcessPoolExecutor類。這些類可以幫助您輕松地創建和管理線程池和進程池，以便在多核處理器上并行執行任務。

以下是一個使用ThreadPoolExecutor進行并發編程性能測試的示例：

import concurrent.futures
import time
import requests

def fetch_url(url):
    response = requests.get(url)
    return response.status_code

urls = [
    "https://www.example.com",
    "https://www.google.com",
    "https://www.github.com",
    # 添加更多URL以進行測試
]

def main():
    with concurrent.futures.ThreadPoolExecutor() as executor:
        start_time = time.time()
        results = list(executor.map(fetch_url, urls))
        end_time = time.time()

    print("Results:", results)
    print(f"Time taken: {end_time - start_time:.2f} seconds")

if __name__ == "__main__":
    main()

在這個示例中，我們定義了一個fetch_url函數，該函數接受一個URL作為參數，并使用requests庫獲取該URL的響應。然后，我們創建了一個urls列表，其中包含要測試的URL。

在main函數中，我們使用ThreadPoolExecutor創建一個線程池，并使用executor.map方法將fetch_url函數應用于urls列表中的每個URL。這將并行執行fetch_url函數，并在完成后返回結果。

最后，我們打印出結果和執行時間。

請注意，ThreadPoolExecutor適用于I/O密集型任務，因為它在等待I/O操作（如網絡請求）完成時會釋放線程。對于CPU密集型任務，可以使用ProcessPoolExecutor來利用多核處理器的優勢。