教學課程：使用 Python 在 Azure 函式中以無伺服器模式將資料發佈至 Socket.IO 用戶端（預覽）

本教學課程會引導您建立與 Azure 函式整合的即時 NASDAQ 索引應用程式，以 Python 的無伺服器模式將資料發佈至 Socket.IO 用戶端。

尋找本教學課程中使用的完整程式碼範例：

Socket.IO 無伺服器 Python 範例

這很重要

預設模式需要永續性伺服器，您無法在預設模式下將適用於 Socket.IO 的 Web PubSub 與 Azure Function 整合。

先決條件

具有有效訂閱的 Azure 帳戶。如果您沒有，可以建立免費帳戶。
Azure 函式核心工具
對 Socket.IO 程式庫有一定的了解。

在無伺服器模式中建立 Web PubSub for Socket.IO 資源

若要建立 Web PubSub for Socket.IO，您可以使用下列 Azure CLI 命令：

az webpubsub create -g <resource-group> -n <resource-name>---kind socketio --service-mode serverless --sku Premium_P1

在本機建立 Azure 函式專案

您應該遵循以下步驟起始本機 Azure Function 專案。

請遵循步驟來安裝最新的 Azure 函式核心工具
在終端機視窗中或從命令提示字元執行下列命令，以在 SocketIOProject 資料夾中建立專案：
```
func init SocketIOProject --worker-runtime python
```
此命令會建立 Python 型函式專案。然後輸入資料夾 SocketIOProject 以執行下列命令。

函式套件組合目前不包含 Socket.IO 函式繫結，因此您必須手動新增套件。

若要排除函式套件組合參考，請編輯 host.json 檔案，並移除下列幾行。

"extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
}

執行命令：

func extensions install -p Microsoft.Azure.WebJobs.Extensions.WebPubSubForSocketIO -v 1.0.0-beta.4

將 function_app.py 中的內容替換為代碼：

import random
import azure.functions as func
from azure.functions.decorators.core import DataType
from azure.functions import Context
import json

app = func.FunctionApp()
current_index= 14000

@app.timer_trigger(schedule="* * * * * *", arg_name="myTimer", run_on_startup=False,
            use_monitor=False)
@app.generic_output_binding("sio", type="socketio", data_type=DataType.STRING, hub="hub")
def publish_data(myTimer: func.TimerRequest,
                sio: func.Out[str]) -> None:
    change = round(random.uniform(-10, 10), 2)
    global current_index
    current_index = current_index + change
    sio.set(json.dumps({
        'actionName': 'sendToNamespace',
        'namespace': '/',
        'eventName': 'update',
        'parameters': [
            current_index
        ]
    }))

@app.function_name(name="negotiate")
@app.route(auth_level=func.AuthLevel.ANONYMOUS)
@app.generic_input_binding("negotiationResult", type="socketionegotiation", hub="hub")
def negotiate(req: func.HttpRequest, negotiationResult) -> func.HttpResponse:
    return func.HttpResponse(negotiationResult)

@app.function_name(name="index")
@app.route(auth_level=func.AuthLevel.ANONYMOUS)
def index(req: func.HttpRequest) -> func.HttpResponse:
    path = './index.html'
    with open(path, 'rb') as f:
        return func.HttpResponse(f.read(), mimetype='text/html')

以下是這些功能的說明：

publish_data：此功能每秒隨機更新納斯達克指數，並透過 Socket.IO 輸出綁定將其廣播給已連接的客戶端。
negotiate：此功能將協商結果回應給客戶端。
index：此函式會傳回靜態 HTML 頁面。

然後新增一個檔案index.html

建立包含以下內容的 index.html 檔案：

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Nasdaq Index</title>
    <style>
        /* Reset some default styles */
        * {
            margin: 0;
            padding: 0;
            box-sizing: border-box;
        }

        body {
            font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
            background: linear-gradient(135deg, #f5f7fa, #c3cfe2);
            height: 100vh;
            display: flex;
            justify-content: center;
            align-items: center;
        }

        .container {
            background-color: white;
            padding: 40px;
            border-radius: 12px;
            box-shadow: 0 4px 6px rgba(0,0,0,0.1);
            text-align: center;
            max-width: 300px;
            width: 100%;
        }

        .nasdaq-title {
            font-size: 2em;
            color: #003087;
            margin-bottom: 20px;
        }

        .index-value {
            font-size: 3em;
            color: #16a34a;
            margin-bottom: 30px;
            transition: color 0.3s ease;
        }

        .update-button {
            padding: 10px 20px;
            font-size: 1em;
            color: white;
            background-color: #003087;
            border: none;
            border-radius: 6px;
            cursor: pointer;
            transition: background-color 0.3s ease;
        }

        .update-button:hover {
            background-color: #002070;
        }
    </style>
</head>
<body>
    <div class="container">
        <div class="nasdaq-title">STOCK INDEX</div>
        <div id="nasdaqIndex" class="index-value">14,000.00</div>
    </div>

    <script src="https://cdn.socket.io/4.7.5/socket.io.min.js"></script>
    <script>
        function updateIndexCore(newIndex) {
            newIndex = parseFloat(newIndex);
            currentIndex = parseFloat(document.getElementById('nasdaqIndex').innerText.replace(/,/g, ''))
            change = newIndex - currentIndex;
            // Update the index value in the DOM
            document.getElementById('nasdaqIndex').innerText = newIndex.toLocaleString('en-US', {minimumFractionDigits: 2, maximumFractionDigits: 2});

            // Optionally, change the color based on increase or decrease
            const indexElement = document.getElementById('nasdaqIndex');
            if (change > 0) {
                indexElement.style.color = '#16a34a'; // Green for increase
            } else if (change < 0) {
                indexElement.style.color = '#dc2626'; // Red for decrease
            } else {
                indexElement.style.color = '#16a34a'; // Neutral color
            }
        }

        async function init() {
            const negotiateResponse = await fetch(`/api/negotiate`);
            if (!negotiateResponse.ok) {
                console.log("Failed to negotiate, status code =", negotiateResponse.status);
                return;
            }
            const negotiateJson = await negotiateResponse.json();
            socket = io(negotiateJson.endpoint, {
                path: negotiateJson.path,
                query: { access_token: negotiateJson.token}
            });

            socket.on('update', (index) => {
                updateIndexCore(index);
            });
        }

        init();
    </script>
</body>
</html>

其中 index.html的關鍵部分：

async function init() {
    const negotiateResponse = await fetch(`/api/negotiate`);
    if (!negotiateResponse.ok) {
        console.log("Failed to negotiate, status code =", negotiateResponse.status);
        return;
    }
    const negotiateJson = await negotiateResponse.json();
    socket = io(negotiateJson.endpoint, {
        path: negotiateJson.path,
        query: { access_token: negotiateJson.token}
    });

    socket.on('update', (index) => {
        updateIndexCore(index);
    });
}

它首先會與函數應用程式交涉，以取得 URI 和指向服務的路徑。並註冊回呼來更新索引。

如何在本機執行應用程式

備妥程式碼之後，請遵循指示來執行範例。

為 Azure Function 設定 Azure 儲存體

Azure Functions 需要儲存體帳戶才能運作，即使在本機執行也是如此。選擇下列兩個選項之一：

執行免費的 Azurite 模擬器。
使用 Azure 儲存體服務。如果您繼續使用它，可能會產生成本。

本機模擬
Azure Blob 儲存服務

安裝 Azurite
```
npm install -g azurite
```

啟動 Azurite 儲存體模擬器：

azurite -l azurite -d azurite\debug.log

確定 AzureWebJobsStorage 中的設定為 UseDevelopmentStorage=true。

更新專案以使用 Azure Blob 儲存體連接字串。

func settings add AzureWebJobsStorage "<storage-connection-string>"

設定 Web PubSub for Socket.IO 的組態

將連接字串新增至函數應用程式：

func settings add WebPubSubForSocketIOConnectionString "<connection string>"

執行範例應用程式

執行通道工具之後，您可以在本機執行函數應用程式：

func start

並造訪位於 http://localhost:7071/api/index 的網頁。

應用程序的屏幕截圖。

後續步驟

接下來，您可以嘗試使用 Bicep，透過身分識別型驗證在線上部署應用程式：

快速入門：在 Socket.IO 無伺服器模式中使用 Azure 函式建置聊天應用程式

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-10-30