提升 Python 應用程式在 Azure Functions 中的吞吐量效能

使用 Python 開發 Azure Functions 時，你需要了解函式的效能，以及這些效能如何影響函式應用的擴展方式。在設計高效能應用程式時，需求更為重要。設計、撰寫及配置功能應用程式時，主要需要考慮的是水平擴展與吞吐量效能配置。

水平調整

預設情況下，Azure Functions 會自動監控你應用程式的負載，並根據需要建立更多 Python 主機實例。 Azure Functions 使用內建的閾值來針對不同的觸發類型決定何時新增實例，例如 QueueTrigger 中的訊息延遲和佇列大小。這些門檻無法使用者自訂。欲了解更多資訊，請參閱 Azure Functions 中的事件驅動尺度。

提升吞吐量效能

預設配置適用於大多數 Azure Functions 應用程式。不過，你可以根據工作負載配置來提升應用程式的吞吐量效能。第一步是了解你正在執行的工作量類型。

工作負載類型	功能應用程式特性	範例
I/O 系結	• 應用程式需要處理多個同時呼叫。 • 應用程式處理大量 I/O 事件，如網路通話及磁碟讀寫。	• 網頁 API
受限於 CPU	• 應用程式執行長時間運算，例如影像調整大小。 • 應用程式進行資料轉換。	• 資料處理 • 機器學習推論

由於真實世界中的函式工作負載通常是 I/O 密集與 CPU 密集的混合，您應在實際的生產負載下分析 App。

性能專屬配置

在你了解功能應用程式的工作負載配置後，以下是你可以用來提升功能吞吐量效能的配置。

非同步
多語言工作者
語言工作程序中的最大數量的工作者
事件迴圈
垂直縮放

非同步

由於 Python 是單執行緒執行時，Python 的主機實例預設一次只能處理一個函式調用。對於處理大量 I/O 事件和/或受 I/O 限制的應用程式，透過非同步執行函式可以大幅提升效能。

若要非同步執行函式，請使用 async def 以下陳述句，直接以 asyncio 執行該函式：

async def main():
    await some_nonblocking_socket_io_op()

這裡有一個使用 aioHTTP HTTP 用戶端的 HTTP 觸發函式範例：

import aiohttp

import azure.functions as func

async def main(req: func.HttpRequest) -> func.HttpResponse:
    async with aiohttp.ClientSession() as client:
        async with client.get("PUT_YOUR_URL_HERE") as response:
            return func.HttpResponse(await response.text())

    return func.HttpResponse(body='NotFound', status_code=404)

沒有關鍵字的 async 函式會自動在 ThreadPoolExecutor 執行緒池中執行：

# Runs in a ThreadPoolExecutor threadpool. Number of threads is defined by PYTHON_THREADPOOL_THREAD_COUNT. 
# The example is intended to show how default synchronous functions are handled.

def main():
    some_blocking_socket_io()

為了充分發揮非同步執行函式的好處，程式碼中使用的 I/O 操作/函式庫也需要實作非同步功能。在定義為非同步的函式中使用同步 I/O 操作可能會損害 整體效能。如果你使用的函式庫沒有實作非同步版本，你仍可能從管理應用程式事件迴圈的非同步執行中受益。

以下是幾個實作非同步模式的客戶端函式庫範例：

aiohttp - 適用於 asyncio 的 Http 客戶端/伺服器
Streams API - 用於網路連線的高階非同步/等待準備原語
Janus Queue - 執行緒安全且可支援asyncio的Python佇列
pyzmq - ZeroMQ 的Python綁定

理解 Python Worker 中的非同步

當你在函式簽名前定義 async 時，Python 會將該函式標記為協程。當您呼叫 Coroutine 時，可將其排程為事件迴圈中的工作。當你在非同步函式中呼叫 await 時，它會將續延註冊到事件迴圈中，讓事件迴圈在等待期間處理下一個任務。

在我們的 Python Worker 中，工作者會與客戶的 async 函式共享事件迴圈，且能同時處理多個請求。我們強烈鼓勵客戶使用相容於 asyncio 的函式庫，如 aiohttp 和 pyzmq。若以同步方式實作，相較於那些程式庫，遵循這些建議可提高函式的輸送量。

備註

如果你的函式被宣告為 async，且實作中沒有任何 await，函式的效能將會受到嚴重影響，因為事件迴圈會被阻塞，Python 工作者無法處理並行請求。

使用多語言工作進程

預設情況下，每個 Functions 主機實例都有一個語言工作程序。你可以透過應用程式 FUNCTIONS_WORKER_PROCESS_COUNT 設定增加每台主機的工作程序數量（最多 10 個）。 Azure Functions 接著嘗試將同時的函式調用平均分配給這些工作者。

對於 CPU 限制的應用程式，你應該將語言工作者數量設為與每個函式應用程式可用核心數相同或更高。欲了解更多，請參閱可用實例 SKU。

I/O 受限的應用程式也可能受益於增加工作程序的數量，超出可用核心數量。請記住，將工作者數量設定過高可能會影響整體效能，因為需要切換上下文的次數增加。

FUNCTIONS_WORKER_PROCESS_COUNT 適用於 Azure Functions 在擴展應用程式以滿足需求時建立的每個主機。

在語言工作者流程中設定最大工作者數量

如非同步章節所述，Python 語言工作者對函數與協程的處理方式不同。 Coroutine 會在與語言工作者相同的事件迴圈中執行。另一方面，函式調用是在 ThreadPoolExecutor 中執行，該執行由語言工作者以執行緒形式維護。

你可以用 PYTHON_THREADPOOL_THREAD_COUNT 應用程式設定設定執行同步函數的最大工人數值。此值設定 ThreadPoolExecutor 物件的 max_worker 參數，允許Python最多使用一個最多max_worker執行緒的執行池來非同步執行呼叫。 PYTHON_THREADPOOL_THREAD_COUNT 適用於函式主機所建立的每個工作者，Python決定何時建立新執行緒或重用現有閒置執行緒。對於較舊的Python版本（即3.8、3.7 和 3.6），max_worker 值設為 1。對於Python版本 3.9，max_worker 設定為 None。

對於 CPU 密集型 App，您應將設定維持在較低的數字，從 1 開始，並隨著工作負載實驗逐步增加。此建議旨在減少切換上下文的時間，並允許 CPU 限制任務完成。

對於 I/O 綁定的應用程式，通過增加每次呼叫中的執行緒數量，應當會帶來顯著的提升。建議先從 Python 預設（核心數）+ 4 開始，然後根據你看到的吞吐量值來調整。

對於混合工作負載的應用程式，你應該同時平衡兩者 FUNCTIONS_WORKER_PROCESS_COUNT 和 PYTHON_THREADPOOL_THREAD_COUNT 設定，以最大化吞吐量。要了解你的函式應用程式花最多時間在哪些方面，我們建議對它們進行分析，並根據其行為設定數值。欲了解這些應用程式設定，請參閱使用多個工作程序。

備註

雖然這些建議適用於 HTTP 和非 HTTP 觸發函式，但你可能需要調整其他觸發特定配置，才能讓函式應用程式達到預期效能。欲了解更多相關資訊，請參閱此 Azure Functions 可靠的最佳實務。

管理事件迴圈

你應該使用相容於 asyncio 的第三方函式庫。如果第三方函式庫都不符合你的需求，你也可以在 Azure Functions 裡管理事件迴圈。管理事件迴圈能讓你在計算資源管理上更有彈性，也能將同步 I/O 函式庫包裝成協程。

有許多有用的 Python 官方文件，使用內建的 asyncio 函式庫討論協程與任務以及事件迴路。

以以下 requests 函式庫為例，這段程式碼片段使用 asyncio 函式庫將 requests.get() 方法包裝成協程，並同時向 SAMPLE_URL 發送多個網頁請求。

import asyncio
import json
import logging

import azure.functions as func
from time import time
from requests import get, Response


async def invoke_get_request(eventloop: asyncio.AbstractEventLoop) -> Response:
    # Wrap requests.get function into a coroutine
    single_result = await eventloop.run_in_executor(
        None,  # using the default executor
        get,  # each task call invoke_get_request
        'SAMPLE_URL'  # the url to be passed into the requests.get function
    )
    return single_result

async def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    eventloop = asyncio.get_event_loop()

    # Create 10 tasks for requests.get synchronous call
    tasks = [
        asyncio.create_task(
            invoke_get_request(eventloop)
        ) for _ in range(10)
    ]

    done_tasks, _ = await asyncio.wait(tasks)
    status_codes = [d.result().status_code for d in done_tasks]

    return func.HttpResponse(body=json.dumps(status_codes),
                             mimetype='application/json')

垂直調整

您也許可以藉由升級至規格更高的進階方案，取得更多處理單位，尤其是在 CPU 密集型作業時。使用較高的處理單元，你可以根據可用核心數量調整工作程序數量，達到更高的平行性。

下一步

欲了解更多Azure Functions Python開發資訊，請參閱以下資源：

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-03-11