Python 並發網路請求與 Timeout 處理最佳實務

– 11 分鐘閱讀

在網頁應用程式開發中，經常需要同時傳送多個網路請求以提升效能。本文將介紹如何使用 Python 的 threading 模組實作並發網路請求，並示範如何設定 Timeout 時間以及處理可能發生的錯誤，最後提供一些最佳實務建議。透過多執行緒，可以讓多個請求同時進行，縮短整體的等待時間。然而，若某些請求耗時過長，可能會影響程式整體的執行效率。因此，設定 Timeout 時間是必要的，可以避免程式因為等待特定請求而卡死。此外，良好的錯誤處理機制可以提升程式的穩定性，避免因為網路錯誤而導致程式當機。最後，我們會討論一些最佳實務，例如遵守網站的服務條款和資料收集政策，以及定期更新程式以適應網站的變更。

並發網路請求的優點

並發網路請求可以讓多個 HTTP 請求同時傳送到伺服器，然後接收回應。這可以大大提高程式的效率，尤其是在需要傳送多個請求的情況下。

使用 threading 模組

以下是使用 threading 模組來實作並發網路請求的範例：

import threading
import requests
import time

def ping(url):
    res = requests.get(url)
    print(f'{url}: {res.text}')

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

start = time.time()
for url in urls:
    ping(url)
print(f'順序執行：{time.time() - start : .2f} 秒')

print()

start = time.time()
threads = []
for url in urls:
    thread = threading.Thread(target=ping, args=(url,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print(f'並發執行：{time.time() - start : .2f} 秒')

在這個範例中，我們使用 threading 模組來建立多個執行緒，每個執行緒負責傳送一個 HTTP 請求。然後，我們使用 start() 方法來啟動每個執行緒，然後使用 join() 方法來等待每個執行緒完成。

結果

執行這個範例後，你會看到並發執行的時間比順序執行的時間短很多。這是因為並發執行可以讓多個請求同時傳送到伺服器，然後接收回應。

圖表翻譯：

  sequenceDiagram
    participant 主程式 as "主程式"
    participant 執行緒1 as "執行緒1"
    participant 執行緒2 as "執行緒2"
    participant 伺服器 as "伺服器"

    Note over 主程式,伺服器: 初始化
    主程式->>執行緒1: 建立執行緒1
    主程式->>執行緒2: 建立執行緒2
    執行緒1->>伺服器: 傳送請求1
    執行緒2->>伺服器: 傳送請求2
    伺服器->>執行緒1: 回應請求1
    伺服器->>執行緒2: 回應請求2
    執行緒1->>主程式: 完成
    執行緒2->>主程式: 完成

這個圖表顯示了並發執行的過程。主程式建立多個執行緒，每個執行緒負責傳送一個 HTTP 請求。然後，執行緒傳送請求到伺服器，然後接收回應。最後，執行緒完成並通知主程式。

平行網路請求

平行網路請求是指在同一時間內傳送多個網路請求，以提高程式的效率。以下是使用 Python 的 threading 模組實作平行網路請求的範例：

import threading
import requests
import time

# 定義一個子執行緒類別
class MyThread(threading.Thread):
    def __init__(self, url):
        threading.Thread.__init__(self)
        self.url = url
        self.result = None

    def run(self):
        res = requests.get(self.url)
        self.result = f'{self.url}: {res.text}'

# 定義 URL 列表
urls = [
    'http://example.com',
    'http://www.google.com',
    'http://www.python.org',
    'http://www.github.com',
    'http://www.stackoverflow.com',
    'http://www.wikipedia.org'
]

# 建立子執行緒列表
threads = []

# 記錄開始時間
start = time.time()

# 建立和啟動子執行緒
for url in urls:
    thread = MyThread(url)
    thread.start()
    threads.append(thread)

# 等待所有子執行緒完成
for thread in threads:
    thread.join()

# 記錄結束時間
print(f'Threading: {time.time() - start : .2f} seconds')

這個範例使用 MyThread 類別定義了一個子執行緒，該子執行緒負責傳送 GET 請求到指定的 URL。然後，程式建立了一個子執行緒列表，將每個 URL 對應的子執行緒新增到列表中。最後，程式啟動所有子執行緒，等待所有子執行緒完成，然後記錄結束時間。

重新設計請求邏輯

為了提高程式的可讀性，我們可以重新設計請求邏輯。以下是重新設計的範例：

import threading
import requests

class MyThread(threading.Thread):
    def __init__(self, url):
        threading.Thread.__init__(self)
        self.url = url
        self.result = None

    def run(self):
        res = requests.get(self.url)
        self.result = f'{self.url}: {res.text}'

def main():
    urls = [
        'http://example.com',
        'http://www.google.com',
        'http://www.python.org',
        'http://www.github.com',
        'http://www.stackoverflow.com',
        'http://www.wikipedia.org'
    ]

    threads = []
    start = time.time()

    for url in urls:
        thread = MyThread(url)
        thread.start()
        threads.append(thread)

    for thread in threads:
        thread.join()

    print(f'Threading: {time.time() - start : .2f} seconds')

if __name__ == '__main__':
    main()

這個範例重新設計了請求邏輯，將其封裝在 MyThread 類別中。然後，程式在 main 函式中建立和啟動子執行緒，等待所有子執行緒完成，然後記錄結束時間。這個設計提高了程式的可讀性和可維護性。

並發網路請求的最佳化

在進行網路請求時，尤其是對多個網站進行請求時，使用並發的方式可以大大提高效率。以下是使用 Python 進行並發網路請求的示例：

import threading
import requests
import time

class MyThread(threading.Thread):
    def __init__(self, url):
        threading.Thread.__init__(self)
        self.url = url
        self.result = None

    def run(self):
        try:
            response = requests.get(self.url, timeout=5)
            self.result = response.status_code
        except requests.exceptions.Timeout:
            self.result = "Timeout"

urls = ["http://httpstat.us/200", "http://httpstat.us/200?sleep=5000", "http://httpstat.us/404"]
threads = [MyThread(url) for url in urls]

start = time.time()

for thread in threads:
    thread.start()

for thread in threads:
    thread.join()

for thread in threads:
    print(f"URL: {thread.url}, Result: {thread.result}")

print(f" Took {time.time() - start : .2f} seconds")
print('Done.')

在這個示例中，我們使用 requests 函式庫進行網路請求，並使用 threading 函式庫進行並發。每個網路請求都在一個單獨的執行緒中進行，這樣可以提高效率。

處理超時

在進行網路請求時，可能會遇到超時的情況。為了處理超時，我們可以使用 requests 函式庫的 timeout 引數。以下是示例：

try:
    response = requests.get(self.url, timeout=5)
    self.result = response.status_code
except requests.exceptions.Timeout:
    self.result = "Timeout"

在這個示例中，如果網路請求超時，則會捕捉 requests.exceptions.Timeout 例外，並將結果設為 “Timeout”。

使用 httpstat.us 進行模擬

httpstat.us 是一個可以用於模擬網路請求的網站。以下是使用 httpstat.us 進行模擬的示例：

urls = ["http://httpstat.us/200", "http://httpstat.us/200?sleep=5000", "http://httpstat.us/404"]

在這個示例中，我們使用 httpstat.us 進行模擬網路請求。其中，http://httpstat.us/200?sleep=5000 會在 5 秒後傳回 200 狀態碼，模擬網路請求的延遲。

結果

最終的結果會顯示每個網路請求的結果，包括狀態碼或超時資訊。以下是示例輸出：

URL: http://httpstat.us/200, Result: 200
URL: http://httpstat.us/200?sleep=5000, Result: Timeout
URL: http://httpstat.us/404, Result: 404
 Took 5.02 seconds
Done.

在這個示例中，第一個網路請求傳回 200 狀態碼，第二個網路請求超時，第三個網路請求傳回 404 狀態碼。總共花費了 5.02 秒。

實作高效的網路請求處理

在進行網路請求的時候，尤其是當我們需要同時傳送多個請求時，如何有效地管理請求的執行時間和回應時間是一個非常重要的問題。下面，我們將探討如何使用 Python 來實作高效的網路請求處理，包括如何設定超時時間和追蹤請求的狀態。

問題描述

當我們同時傳送多個網路請求時，如果其中一個請求需要很長時間才能傳回回應，可能會導致整個程式的執行時間大大增加。為了避免這種情況，我們需要實作一個機制，可以在請求超過一定時間後自動判斷為超時，並傳回相應的結果。

解決方案

為了實作高效的網路請求處理，我們可以使用 Python 的 threading 模組來建立多個執行緒，同時傳送多個請求。同時，我們需要設定一個超時時間，當請求超過這個時間後，就會被判斷為超時。

import requests
import threading
import time

class MyThread(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = None

    def run(self):
        try:
            res = requests.get(self.url, timeout=5)  # 設定超時時間為 5 秒
            self.result = f'{self.url}: {res.text}'
        except requests.Timeout:
            self.result = f'{self.url}: Timeout'

def process_requests(threads, timeout=5):
    start_time = time.time()
    while True:
        alive_count = sum(1 for thread in threads if thread.is_alive())
        if alive_count == 0 or time.time() - start_time > timeout:
            break
        time.sleep(0.01)  # 每 10 毫秒檢查一次

    for thread in threads:
        if thread.result:
            print(thread.result)

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

threads = [MyThread(url) for url in urls]
for thread in threads:
    thread.start()

process_requests(threads, timeout=10)

在上面的程式碼中，我們定義了一個 MyThread 類，繼承自 threading.Thread。在 run 方法中，我們使用 requests.get 傳送 GET 請求，並設定超時時間為 5 秒。如果請求超時，就會捕捉 requests.Timeout 異常，並傳回相應的結果。

在 process_requests 函式中，我們使用一個 while 迴圈不斷檢查執行緒的狀態，如果所有執行緒都完成或超過設定的超時時間，就會離開迴圈並列印預出所有執行緒的結果。

並發網路請求處理

在進行網路請求的過程中，如何有效地管理和處理多個請求是一個重要的挑戰。為瞭解決這個問題，我們可以使用多執行緒（multithreading）技術來實作並發網路請求。

多執行緒基礎

Python 的 threading 模組提供了基本的多執行緒支援。下面是一個簡單的例子，展示如何使用 threading 來建立和管理執行緒：

import threading
import requests

class MyThread(threading.Thread):
    def __init__(self, url):
        threading.Thread.__init__(self)
        self.url = url
        self.result = f'{self.url}: Custom timeout'

    def run(self):
        res = requests.get(self.url)
        # 處理請求結果
        self.result = res.text

# 建立多個執行緒
threads = []
urls = ['http://example.com', 'http://example.org', 'http://example.net']
for url in urls:
    thread = MyThread(url)
    thread.start()
    threads.append(thread)

# 等待所有執行緒完成
while any(thread.is_alive() for thread in threads):
    pass

# 列印請求結果
for thread in threads:
    print(thread.result)

進階多執行緒管理

在上面的例子中，我們簡單地建立了多個執行緒並等待它們完成。但是在實際應用中，我們可能需要更進一步的控制，例如設定超時時間、處理異常等。下面是一個更進一步的例子：

import threading
import requests

class MyThread(threading.Thread):
    def __init__(self, url):
        threading.Thread.__init__(self)
        self.url = url
        self.result = f'{self.url}: Custom timeout'

    def run(self):
        try:
            res = requests.get(self.url, timeout=5)
            self.result = res.text
        except requests.exceptions.Timeout:
            self.result = f'{self.url}: Timeout'

def process_requests(threads, timeout=10):
    def alive_count():
        return sum(1 for thread in threads if thread.is_alive())

    while alive_count() > 0 and timeout > 0:
        timeout -= 1
        threading.sleep(1)
        for thread in threads:
            if not thread.is_alive():
                print(thread.result)

# 建立多個執行緒
threads = []
urls = ['http://example.com', 'http://example.org', 'http://example.net']
for url in urls:
    thread = MyThread(url)
    thread.start()
    threads.append(thread)

# 處理請求
process_requests(threads, timeout=10)

在這個例子中，我們定義了一個 process_requests 函式，負責管理多個執行緒的執行和超時控制。函式內部定義了一個 alive_count 函式，傳回目前還在執行的執行緒數量。然後，函式使用一個 while 迴圈不斷檢查執行緒的狀態和超時時間，直到所有執行緒完成或超時時間到期。

多執行緒與Timeout處理

在多執行緒的應用中，處理Timeout是一個非常重要的議題。下面是一個使用Python的多執行緒例項，展示瞭如何處理Timeout。

問題描述

假設我們有多個URL需要存取，但是有些URL的回應時間可能很長，甚至超過我們的耐心。為了避免程式卡死，我們需要設定一個Timeout時間，當Timeout時間到達時，程式就會繼續執行，不再等待那些回應時間太長的URL。

程式碼實作

import threading
import time
import requests

class MyThread(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = f'{self.url}: Timeout'

    def run(self):
        try:
            res = requests.get(self.url, timeout=10)
            self.result = f'{self.url}: {res.text}'
        except requests.exceptions.RequestException as e:
            self.result = f'{self.url}: {e}'

def process_requests(threads):
    timeout = 5
    while threading.active_count() > 1 and timeout > 0:
        timeout -= 0.1
        time.sleep(0.1)
    for thread in threads:
        print(thread.result)

urls = [
    'http://example.com',
    'http://example.com/delay/4',
    'http://example.com/delay/20'
]

start = time.time()

threads = [MyThread(url) for url in urls]
for thread in threads:
    thread.setDaemon(True)
    thread.start()

process_requests(threads)
print(f'Took {time.time() - start : .2f} seconds')
print('Done.')

執行結果

http://example.com: <html>...</html>
http://example.com/delay/4: <html>...</html>
http://example.com/delay/20: Timeout
Took 5.70 seconds
Done.

分析

在這個例項中，我們設定了一個Timeout時間為5秒。當Timeout時間到達時，程式就會繼續執行，不再等待那些回應時間太長的URL。透過設定daemon thread，我們可以讓主程式在Timeout時間到達時就離開，即使有些thread還在執行。

並發式網路請求最佳實踐

在進行並發式網路請求時，需要謹慎考慮和實施多個方面。以下將討論這些方面和最佳實踐，以幫助您開發更好的應用程式。

考慮服務條款和資料收集政策

未經授權的資料收集已成為科技界近年來的熱門話題。因此，開發人員在進行自動化網路請求時，應該仔細檢視網站的服務條款和資料收集政策。這些政策通常可以在網站的服務條款或類似檔案中找到。如果有疑問，通常最好直接聯絡網站以詢問更多詳細資訊。

錯誤處理

錯誤是程式設計中不可避免的，尤其是在進行網路請求時。這些錯誤可能包括傳送無效請求、網路連線問題、下載的HTML程式碼錯誤或HTML程式碼解析失敗。因此，使用Python的try…except區塊和其他錯誤處理工具來避免應用程式當機是非常重要的。避免當機尤其重要，如果您的程式碼或應用程式用於生產環境或大型應用程式中。

在並發式網路爬蟲中，可能有一些執行緒成功收集資料，而其他執行緒失敗。透過使用try…except區塊，您可以確保失敗的執行緒不會導致整個程式當機，並且成功的執行緒仍然可以傳回結果。

然而，盲目地捕捉錯誤並不理想。這種做法是指在程式中使用一個大型的try…except區塊來捕捉所有錯誤，但無法獲得更多有關錯誤的資訊。這種做法也被稱為錯誤吞噬。建議您在程式中使用具體的錯誤處理程式碼，以便對特定的錯誤採取適當的行動，並可能揭示其他未考慮到的錯誤。

定期更新程式

網站經常更改其請求處理邏輯和顯示的資料。 如果一個程式與網站伺服器互動的邏輯相當僵化（例如，結構化請求、只處理一種回應等），那麼當網站修改其請求處理邏輯時，程式很可能會停止正常運作。這種情況經常發生在網路爬蟲程式中，當HTML標籤改變時，這些程式將無法找到其資料。

為了防止自動化資料收集程式運作，網站可能會實施這種做法。唯一繼續使用已更改其請求處理邏輯的網站的方法是分析更新的協定並相應地修改程式。

程式碼範例

以下是使用Python進行並發式網路請求的範例，展示瞭如何使用try…except區塊和具體的錯誤處理程式碼：

import requests
from concurrent.futures import ThreadPoolExecutor

def fetch_url(url):
    try:
        response = requests.get(url)
        response.raise_for_status()  # 引發異常如果HTTP請求傳回了4XX/5XX狀態碼
        return response.text
    except requests.RequestException as e:
        print(f"錯誤發生：{e}")
        return None

def main():
    urls = ["http://example.com/page1", "http://example.com/page2", "http://example.com/page3"]
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(fetch_url, url): url for url in urls}
        for future in futures:
            url = futures[future]
            try:
                data = future.result()
                if data is not None:
                    print(f"成功從{url}收集資料：{data}")
            except Exception as e:
                print(f"錯誤發生：{e}")

if __name__ == "__main__":
    main()

這個範例使用requests函式庫傳送HTTP請求，並使用concurrent.futures函式庫進行並發式請求。fetch_url函式使用try…except區塊捕捉任何發生的錯誤，並傳回None如果錯誤發生。main函式使用ThreadPoolExecutor建立一個執行緒池，並提交每個URL的請求。然後，它等待每個請求完成，並列印收集到的資料或錯誤資訊。

什麼是網頁請求和HTTP協定？

網頁請求是指使用者端（如網頁瀏覽器）向伺服器傳送請求，以取得網頁資源的過程。HTTP（超文字傳輸協定）是一種用於網頁請求的協定，它定義了使用者端和伺服器之間的溝通規則。

從效能最佳化視角來看，並發網路請求能顯著提升應用程式效率，尤其在處理大量網路請求時效果更為突出。本文分析了 Python 中使用 threading 模組及 concurrent.futures 函式庫實作並發請求的各種方法，並探討了超時處理、錯誤處理等關鍵議題。然而，單純追求速度並非最佳方案，需考量伺服器端負載及網站服務條款限制。技術限制深析顯示，多執行緒在 Python 中受全域性直譯器鎖（GIL）影響，並非真正的平行，對於 CPU 密集型任務提升有限。

展望未來，非同步程式設計模型（例如 asyncio）和多程式方案將成為處理 I/O 密集型任務的主流方向，更能發揮多核心處理器的效能優勢。同時，更精細化的錯誤處理和請求管理策略，例如請求重試機制和熔斷機制，也將成為提升網路請求穩定性和可靠性的關鍵。玄貓認為，開發者應根據實際應用場景選擇合適的並發策略，並持續關注相關技術的發展趨勢，才能在兼顧效能與穩健性的前提下，構建高效的網路應用程式。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。