你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

快速入门：采用 Python 通过 Azure Functions 和 Azure SignalR 服务创建无服务器应用

项目
04/28/2024

使用 Azure Functions 和 Python 生成将消息广播到客户端的无服务器应用程序，以开始使用 Azure SignalR 服务。你将在本地环境中运行函数，并连接到云中的 Azure SignalR 服务实例。完成本快速入门会从你的 Azure 帐户中扣取最多几美分的费用。

注意

可以从 GitHub 获取本文中的代码。

先决条件

本快速入门可以在 macOS、Windows 或 Linux 上运行。需要以下各项：

先决条件	说明
Azure 订阅	如果没有 Azure 订阅，可创建一个 Azure 免费帐户
代码编辑器	你需要一个代码编辑器，如 Visual Studio Code。
Azure Functions Core Tools	需要版本 2.7.1505 或更高版本才能在本地运行 Python Azure 函数应用。
Python 3.7+	Azure Functions 需要 Python 3.7+。请参阅支持的 Python 版本。
Azurite	SignalR 绑定需要 Azure 存储。在本地运行函数时，可以使用本地存储仿真器。
Azure CLI	（可选）可以使用 Azure CLI 创建 Azure SignalR 服务实例。

创建 Azure SignalR 服务实例

在本部分中，你将创建一个基本 Azure SignalR 实例来用于你的应用。以下步骤使用 Azure 门户创建新实例，但你也可以使用 Azure CLI。有关详细信息，请参阅 Azure SignalR 服务 CLI 参考中的 az signalr create 命令。

登录 Azure 门户。
在页面的左上角，选择“+ 创建资源” 。
在“创建资源”页上，在“搜索服务和市场”文本框中，输入“signalr”，然后从列表中选择“SignalR 服务”。
在“SignalR 服务”页上，选择“创建”。
在“基本信息”选项卡上，输入新 SignalR 服务实例的基本信息。输入以下值：

字段	建议的值	描述
订阅	选择订阅	选择要用于创建新的 SignalR 服务实例的订阅。
资源组	创建一个名为 SignalRTestResources 的资源组	为 SignalR 资源选择或创建资源组。对于本教程，创建新的资源组比使用现有资源组更为合适。若要在完成本教程后释放资源，请删除资源组。删除资源组还会删除属于该组的所有资源。此操作不能撤消。删除资源组之前，请确保它不包含你希望保留的资源。有关详细信息，请参阅 Using resource groups to manage your Azure resources（使用资源组管理 Azure 资源）。
资源名称	testsignalr	输入用于 SignalR 资源的唯一资源名称。如果你的区域中已使用了 testsignalr，请添加一个数字或字符，以将名称设为唯一。该名称必须是包含 1 到 63 个字符的字符串，只能包含数字、字母和连字符 (`-`) 字符。该名称的开头或末尾不能是连字符字符，并且连续的连字符字符无效。
区域	选择你的区域	为新的 SignalR 服务实例选择合适的区域。 Azure SignalR 服务当前并非在所有区域中都可用。有关详细信息，请参阅 Azure SignalR 服务区域可用性。
定价层	选择“更改”，然后选择“免费(仅限开发/测试)”。选择“选择”以确认你选择的定价层。	Azure SignalR 服务有三个定价层：免费、标准和高级。教程使用的是免费层，除非在先决条件中另行说明。若要详细了解各个层级之间的功能差异以及定价，请参阅 Azure SignalR 服务定价
服务模式	选择适当的服务模式	在 Web 应用中托管 SignalR 中心逻辑并使用 SignalR 服务作为代理时，请使用“默认”。使用无服务器技术（如 Azure Functions）托管 SignalR 中心逻辑时，请使用“无服务器”。 “经典”模式仅用于向后兼容，不建议使用。有关详细信息，请参阅 Azure SignalR 服务中的服务模式。

对于 SignalR 教程，你不需要更改“网络”和“标记”选项卡上的设置。

选择“基本信息”选项卡底部的“查看 + 创建”按钮。
在“查看 + 创建”选项卡上检查各个值，然后选择“创建”。部署需要几分钟时间才能完成。
在部署完成后，选择“转到资源组”按钮。
在 SignalR 资源页面上，从左侧菜单中选择“设置”下的“密钥”。
复制主密钥的连接字符串。在本教程中，稍后你将需要使用此连接字符串来配置你的应用。

创建 Azure 函数项目

创建本地 Azure 函数项目。

通过命令行创建项目的目录。
更改为项目目录。
使用 Azure Functions func init 命令初始化函数项目。

# Initialize a function project
func init --worker-runtime python

创建函数

初始化项目后，需要创建函数。此项目需要三个函数：

index：托管客户端的网页。
negotiate：允许客户端获取访问令牌。
broadcast：使用时间触发器定期将消息广播到所有客户端。

从项目的根目录运行 func new 命令时，Azure Functions Core Tools 会在 function_app.py 文件中追加函数代码。你将根据需要编辑参数和内容，方法是将默认代码替换为应用代码。

创建索引函数

可将此示例函数用作自己的函数的模板。

打开 function_app.py 文件。此文件将包含你的函数。首先，修改该文件以包含必要的导入语句，并定义我们将在以下函数中使用的全局变量。

import azure.functions as func
import os
import requests
import json 

app = func.FunctionApp()

etag = ''
start_count = 0

通过添加以下代码添加函数 index

@app.route(route="index", auth_level=func.AuthLevel.ANONYMOUS)
def index(req: func.HttpRequest) -> func.HttpResponse:
    f = open(os.path.dirname(os.path.realpath(__file__)) + '/content/index.html')
    return func.HttpResponse(f.read(), mimetype='text/html')

此函数会托管客户端的网页。

创建协商函数

通过添加以下代码添加函数 negotiate

@app.route(route="negotiate", auth_level=func.AuthLevel.ANONYMOUS, methods=["POST"])
@app.generic_input_binding(arg_name="connectionInfo", type="signalRConnectionInfo", hubName="serverless", connectionStringSetting="AzureSignalRConnectionString")
def negotiate(req: func.HttpRequest, connectionInfo) -> func.HttpResponse:
    return func.HttpResponse(connectionInfo)

此函数允许客户端获取访问令牌。

创建广播函数。

通过添加以下代码添加函数 broadcast

@app.timer_trigger(schedule="*/1 * * * *", arg_name="myTimer",
              run_on_startup=False,
              use_monitor=False)
@app.generic_output_binding(arg_name="signalRMessages", type="signalR", hubName="serverless", connectionStringSetting="AzureSignalRConnectionString")
def broadcast(myTimer: func.TimerRequest, signalRMessages: func.Out[str]) -> None:
    global etag
    global start_count
    headers = {'User-Agent': 'serverless', 'If-None-Match': etag}
    res = requests.get('https://api.github.com/repos/azure/azure-functions-python-worker', headers=headers)
    if res.headers.get('ETag'):
        etag = res.headers.get('ETag')

    if res.status_code == 200:
        jres = res.json()
        start_count = jres['stargazers_count']

    signalRMessages.set(json.dumps({
        'target': 'newMessage',
        'arguments': [ 'Current star count of https://api.github.com/repos/azure/azure-functions-python-worker is: ' + str(start_count) ]
    }))

此函数使用时间触发器定期将消息广播到所有客户端。

创建 Azure 函数项目

创建本地 Azure 函数项目。

通过命令行创建项目的目录。
更改为项目目录。
使用 Azure Functions func init 命令初始化函数项目。

# Initialize a function project
func init --worker-runtime python --model v1

创建函数

初始化项目后，需要创建函数。此项目需要三个函数：

index：托管客户端的网页。
negotiate：允许客户端获取访问令牌。
broadcast：使用时间触发器定期将消息广播到所有客户端。

从项目的根目录运行 func new 命令时，Azure Functions Core Tools 会创建默认的函数源文件，并将其存储在以函数命名的文件夹中。根据需要编辑文件，将默认代码替换为应用代码。

创建索引函数

可将此示例函数用作自己的函数的模板。

运行以下命令来创建 index 函数。

func new -n index -t HttpTrigger

编辑 index/function.json，并将内容替换为以下 json 代码：

{
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

编辑 index/_init_.py 并将内容替换为以下代码：

import os

import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
    f = open(os.path.dirname(os.path.realpath(__file__)) + '/../content/index.html')
    return func.HttpResponse(f.read(), mimetype='text/html')

创建协商函数

运行以下命令来创建 negotiate 函数。

func new -n negotiate -t HttpTrigger

编辑 negotiate/function.json，并将内容替换为以下 json 代码：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    },
    {
      "type": "signalRConnectionInfo",
      "name": "connectionInfo",
      "hubName": "serverless",
      "connectionStringSetting": "AzureSignalRConnectionString",
      "direction": "in"
    }
  ]
}

编辑 negotiate/_init_.py 并将内容替换为以下代码：

import azure.functions as func


def main(req: func.HttpRequest, connectionInfo) -> func.HttpResponse:
    return func.HttpResponse(connectionInfo)

创建广播函数。

运行以下命令来创建 broadcast 函数。

func new -n broadcast -t TimerTrigger
# install requests
pip install requests

编辑 broadcast/function.json，并将内容替换为以下代码：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "myTimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "*/5 * * * * *"
    },
    {
      "type": "signalR",
      "name": "signalRMessages",
      "hubName": "serverless",
      "connectionStringSetting": "AzureSignalRConnectionString",
      "direction": "out"
    }
  ]
}

编辑 broadcast/_init_.py，并将内容替换为以下代码：

import requests
import json

import azure.functions as func

etag = ''
start_count = 0

def main(myTimer: func.TimerRequest, signalRMessages: func.Out[str]) -> None:
    global etag
    global start_count
    headers = {'User-Agent': 'serverless', 'If-None-Match': etag}
    res = requests.get('https://api.github.com/repos/azure/azure-signalr', headers=headers)
    if res.headers.get('ETag'):
        etag = res.headers.get('ETag')

    if res.status_code == 200:
        jres = res.json()
        start_count = jres['stargazers_count']

    signalRMessages.set(json.dumps({
        'target': 'newMessage',
        'arguments': [ 'Current star count of https://github.com/Azure/azure-signalr is: ' + str(start_count) ]
    }))

创建 index.html 文件

本应用的客户端界面是网页。 index 函数从 content/index.html 文件中读取 HTML 内容。

在项目根文件夹中创建名为 content 的文件夹。
创建 content/index.html 文件。
将以下内容复制到 content/index.html 文件并保存它：

<html>

<body>
  <h1>Azure SignalR Serverless Sample</h1>
  <div id="messages"></div>
  <script src="https://cdnjs.cloudflare.com/ajax/libs/microsoft-signalr/3.1.7/signalr.min.js"></script>
  <script>
    let messages = document.querySelector('#messages');
    const apiBaseUrl = window.location.origin;
    const connection = new signalR.HubConnectionBuilder()
        .withUrl(apiBaseUrl + '/api')
        .configureLogging(signalR.LogLevel.Information)
        .build();
      connection.on('newMessage', (message) => {
        document.getElementById("messages").innerHTML = message;
      });

      connection.start()
        .catch(console.error);
  </script>
</body>

</html>

将 SignalR 服务的连接字符串添加到函数应用设置

最后一步是在 Azure 函数应用设置中设置 SignalR 服务连接字符串。

在 Azure 门户中，转到之前部署的 SignalR 实例。
选择“密钥”以查看 SignalR 服务实例的连接字符串。

复制主连接字符串并执行命令：

func settings add AzureSignalRConnectionString "<signalr-connection-string>"

在本地运行 Azure 函数应用

启动 Azurite 存储模拟器：

azurite

在本地环境中运行 Azure 函数应用：

func start

注意

如果在 Blob 存储上看到一个错误显示读取错误，请确保将 local.settings.json 文件中的“AzureWebJobsStorage”设置设为 UseDevelopmentStorage=true。

在本地运行 Azure 函数后，转到 http://localhost:7071/api/index。该页会显示 GitHub Azure/azure-signalr 存储库的当前星级数。如果在 GitHub 中为存储库添加星标或取消添加星标，每隔几秒就会看到刷新的星数。

清理资源

如果不打算继续使用此应用，请按照以下步骤删除本快速入门中创建的所有资源，以免产生任何费用：

在 Azure 门户的最左侧选择“资源组”，，然后选择创建的资源组。或者，可以使用搜索框按名称查找资源组。
在打开的窗口中选择资源组，然后单击“删除资源组”。
在新窗口中键入要删除的资源组的名称，然后单击“删除”。

遇到问题？尝试使用故障排除指南或通知我们。

后续步骤

在本快速入门中，你在本地生成并运行了一个实时的无服务器应用程序。接着，将详细了解如何通过 SignalR 服务在客户端与 Azure Functions 之间进行双向通信。

Azure Functions 的 SignalR Service 绑定

无服务器双向通信

使用 VS Code 部署 Azure Functions

通过

快速入门：采用 Python 通过 Azure Functions 和 Azure SignalR 服务创建无服务器应用

先决条件

创建 Azure SignalR 服务实例

创建 Azure 函数项目

创建函数

创建索引函数

创建协商函数

创建广播函数。

创建 Azure 函数项目

创建函数

创建索引函数

创建协商函数

创建广播函数。

创建 index.html 文件

将 SignalR 服务的连接字符串添加到函数应用设置

在本地运行 Azure 函数应用

清理资源

后续步骤

反馈

其他资源