Azure Functions 的 Azure 服务总线触发器

使用服务总线触发器响应来自服务总线队列或主题的消息。 从扩展版本 3.1.0 开始,可以在启用会话的队列或主题上触发。

有关设置和配置详细信息,请参阅概述

消耗计划和高级计划的服务总线缩放决策取决于基于目标的缩放。 有关详细信息,请参阅基于目标的缩放

重要

本文使用选项卡来支持多个版本的 Node.js 编程模型。 v4 模型已正式发布,旨在为 JavaScript 和 TypeScript 开发人员提供更为灵活和直观的体验。 有关 v4 模型工作原理的更多详细信息,请参阅 Azure Functions Node.js 开发人员指南。 要详细了解 v3 和 v4 之间的差异,请参阅迁移指南

Azure Functions 支持两种 Python 编程模型。 定义绑定的方式取决于选择的编程模型。

使用 Python v2 编程模型,可以直接在 Python 函数代码中使用修饰器定义绑定。 有关详细信息,请参阅 Python 开发人员指南

本文同时支持两个编程模型。

示例

可使用以下 C# 模式之一来创建 C# 函数:

  • 独立辅助角色模型:编译的 C# 函数,该函数在独立于运行时的工作进程中运行。 需要独立工作进程才能支持在 LTS 和非 LTS 版 .NET 和 .NET Framework 上运行的 C# 函数。 独立工作进程函数的扩展使用 Microsoft.Azure.Functions.Worker.Extensions.* 命名空间。
  • 进程内模型:编译的 C# 函数,该函数在与 Functions 运行时相同的进程中运行。 在此模型的变体中,可以使用 C# 脚本运行 Functions,该脚本主要用于 C# 门户编辑。 进程内函数的扩展使用 Microsoft.Azure.WebJobs.Extensions.* 命名空间。

此代码定义并初始化 ILogger

// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using System.Threading.Tasks;
using Azure.Messaging.ServiceBus;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

namespace SampleApp
{
    /// <summary>
    /// Samples demonstrating binding to the <see cref="ServiceBusReceivedMessage"/> type.
    /// </summary>
    public class ServiceBusReceivedMessageFunctions
    {
        //<docsnippet_servicebusmessage_createlogger>
        private readonly ILogger<ServiceBusReceivedMessageFunctions> _logger;

        public ServiceBusReceivedMessageFunctions(ILogger<ServiceBusReceivedMessageFunctions> logger)
        {
            _logger = logger;
        }
        //</docsnippet_servicebusmessage_createlogger>
        /// <summary>
        /// This function demonstrates binding to a single <see cref="ServiceBusReceivedMessage"/>.
        /// </summary>
        //<docsnippet_servicebus_readmessage>
        [Function(nameof(ServiceBusReceivedMessageFunction))]
        [ServiceBusOutput("outputQueue", Connection = "ServiceBusConnection")]
        public string ServiceBusReceivedMessageFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection")] ServiceBusReceivedMessage message)
        {
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message Body: {body}", message.Body);
            _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);

            var outputMessage = $"Output message created at {DateTime.Now}";
            return outputMessage;
        }
        //</docsnippet_servicebus_readmessage>
        /// <summary>
        /// This function demonstrates binding to an array of <see cref="ServiceBusReceivedMessage"/>.
        /// Note that when doing so, you must also set the <see cref="ServiceBusTriggerAttribute.IsBatched"/> property
        /// to <value>true</value>.
        /// </summary>
        //<docsnippet_servicebus_readbatch>
        [Function(nameof(ServiceBusReceivedMessageBatchFunction))]
        public void ServiceBusReceivedMessageBatchFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection", IsBatched = true)] ServiceBusReceivedMessage[] messages)
        {
            foreach (ServiceBusReceivedMessage message in messages)
            {
                _logger.LogInformation("Message ID: {id}", message.MessageId);
                _logger.LogInformation("Message Body: {body}", message.Body);
                _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);
            }
        }
        //</docsnippet_servicebus_readbatch>
        /// <summary>
        /// This functions demonstrates that it is possible to bind to both the ServiceBusReceivedMessage and any of the supported binding contract
        /// properties at the same time. If attempting this, the ServiceBusReceivedMessage must be the first parameter. There is not
        /// much benefit to doing this as all of the binding contract properties are available as properties on the ServiceBusReceivedMessage.
        /// </summary>
        [Function(nameof(ServiceBusReceivedMessageWithStringProperties))]
        public void ServiceBusReceivedMessageWithStringProperties(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection")]
            ServiceBusReceivedMessage message, string messageId, int deliveryCount)
        {
            // The MessageId property and the messageId parameter are the same.
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message ID: {id}", messageId);

            // Similarly the DeliveryCount property and the deliveryCount parameter are the same.
            _logger.LogInformation("Delivery Count: {count}", message.DeliveryCount);
            _logger.LogInformation("Delivery Count: {count}", deliveryCount);
        }
        //<docsnippet_servicebus_message_actions>
        [Function(nameof(ServiceBusMessageActionsFunction))]
        public async Task ServiceBusMessageActionsFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection", AutoCompleteMessages = false)]
            ServiceBusReceivedMessage message,
            ServiceBusMessageActions messageActions)
        {
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message Body: {body}", message.Body);
            _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);

            // Complete the message
            await messageActions.CompleteMessageAsync(message);
        }
        //</docsnippet_servicebus_message_actions>
    }
}

此示例演示接收单个服务总线队列消息并将其写入日志的 C# 函数

// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using System.Threading.Tasks;
using Azure.Messaging.ServiceBus;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

namespace SampleApp
{
    /// <summary>
    /// Samples demonstrating binding to the <see cref="ServiceBusReceivedMessage"/> type.
    /// </summary>
    public class ServiceBusReceivedMessageFunctions
    {
        //<docsnippet_servicebusmessage_createlogger>
        private readonly ILogger<ServiceBusReceivedMessageFunctions> _logger;

        public ServiceBusReceivedMessageFunctions(ILogger<ServiceBusReceivedMessageFunctions> logger)
        {
            _logger = logger;
        }
        //</docsnippet_servicebusmessage_createlogger>
        /// <summary>
        /// This function demonstrates binding to a single <see cref="ServiceBusReceivedMessage"/>.
        /// </summary>
        //<docsnippet_servicebus_readmessage>
        [Function(nameof(ServiceBusReceivedMessageFunction))]
        [ServiceBusOutput("outputQueue", Connection = "ServiceBusConnection")]
        public string ServiceBusReceivedMessageFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection")] ServiceBusReceivedMessage message)
        {
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message Body: {body}", message.Body);
            _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);

            var outputMessage = $"Output message created at {DateTime.Now}";
            return outputMessage;
        }
        //</docsnippet_servicebus_readmessage>
        /// <summary>
        /// This function demonstrates binding to an array of <see cref="ServiceBusReceivedMessage"/>.
        /// Note that when doing so, you must also set the <see cref="ServiceBusTriggerAttribute.IsBatched"/> property
        /// to <value>true</value>.
        /// </summary>
        //<docsnippet_servicebus_readbatch>
        [Function(nameof(ServiceBusReceivedMessageBatchFunction))]
        public void ServiceBusReceivedMessageBatchFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection", IsBatched = true)] ServiceBusReceivedMessage[] messages)
        {
            foreach (ServiceBusReceivedMessage message in messages)
            {
                _logger.LogInformation("Message ID: {id}", message.MessageId);
                _logger.LogInformation("Message Body: {body}", message.Body);
                _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);
            }
        }
        //</docsnippet_servicebus_readbatch>
        /// <summary>
        /// This functions demonstrates that it is possible to bind to both the ServiceBusReceivedMessage and any of the supported binding contract
        /// properties at the same time. If attempting this, the ServiceBusReceivedMessage must be the first parameter. There is not
        /// much benefit to doing this as all of the binding contract properties are available as properties on the ServiceBusReceivedMessage.
        /// </summary>
        [Function(nameof(ServiceBusReceivedMessageWithStringProperties))]
        public void ServiceBusReceivedMessageWithStringProperties(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection")]
            ServiceBusReceivedMessage message, string messageId, int deliveryCount)
        {
            // The MessageId property and the messageId parameter are the same.
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message ID: {id}", messageId);

            // Similarly the DeliveryCount property and the deliveryCount parameter are the same.
            _logger.LogInformation("Delivery Count: {count}", message.DeliveryCount);
            _logger.LogInformation("Delivery Count: {count}", deliveryCount);
        }
        //<docsnippet_servicebus_message_actions>
        [Function(nameof(ServiceBusMessageActionsFunction))]
        public async Task ServiceBusMessageActionsFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection", AutoCompleteMessages = false)]
            ServiceBusReceivedMessage message,
            ServiceBusMessageActions messageActions)
        {
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message Body: {body}", message.Body);
            _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);

            // Complete the message
            await messageActions.CompleteMessageAsync(message);
        }
        //</docsnippet_servicebus_message_actions>
    }
}

此示例演示一个 C# 函数,该函数接收单个批处理中的多个服务总线队列消息,并将每个消息写入日志:

// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using System.Threading.Tasks;
using Azure.Messaging.ServiceBus;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

namespace SampleApp
{
    /// <summary>
    /// Samples demonstrating binding to the <see cref="ServiceBusReceivedMessage"/> type.
    /// </summary>
    public class ServiceBusReceivedMessageFunctions
    {
        //<docsnippet_servicebusmessage_createlogger>
        private readonly ILogger<ServiceBusReceivedMessageFunctions> _logger;

        public ServiceBusReceivedMessageFunctions(ILogger<ServiceBusReceivedMessageFunctions> logger)
        {
            _logger = logger;
        }
        //</docsnippet_servicebusmessage_createlogger>
        /// <summary>
        /// This function demonstrates binding to a single <see cref="ServiceBusReceivedMessage"/>.
        /// </summary>
        //<docsnippet_servicebus_readmessage>
        [Function(nameof(ServiceBusReceivedMessageFunction))]
        [ServiceBusOutput("outputQueue", Connection = "ServiceBusConnection")]
        public string ServiceBusReceivedMessageFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection")] ServiceBusReceivedMessage message)
        {
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message Body: {body}", message.Body);
            _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);

            var outputMessage = $"Output message created at {DateTime.Now}";
            return outputMessage;
        }
        //</docsnippet_servicebus_readmessage>
        /// <summary>
        /// This function demonstrates binding to an array of <see cref="ServiceBusReceivedMessage"/>.
        /// Note that when doing so, you must also set the <see cref="ServiceBusTriggerAttribute.IsBatched"/> property
        /// to <value>true</value>.
        /// </summary>
        //<docsnippet_servicebus_readbatch>
        [Function(nameof(ServiceBusReceivedMessageBatchFunction))]
        public void ServiceBusReceivedMessageBatchFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection", IsBatched = true)] ServiceBusReceivedMessage[] messages)
        {
            foreach (ServiceBusReceivedMessage message in messages)
            {
                _logger.LogInformation("Message ID: {id}", message.MessageId);
                _logger.LogInformation("Message Body: {body}", message.Body);
                _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);
            }
        }
        //</docsnippet_servicebus_readbatch>
        /// <summary>
        /// This functions demonstrates that it is possible to bind to both the ServiceBusReceivedMessage and any of the supported binding contract
        /// properties at the same time. If attempting this, the ServiceBusReceivedMessage must be the first parameter. There is not
        /// much benefit to doing this as all of the binding contract properties are available as properties on the ServiceBusReceivedMessage.
        /// </summary>
        [Function(nameof(ServiceBusReceivedMessageWithStringProperties))]
        public void ServiceBusReceivedMessageWithStringProperties(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection")]
            ServiceBusReceivedMessage message, string messageId, int deliveryCount)
        {
            // The MessageId property and the messageId parameter are the same.
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message ID: {id}", messageId);

            // Similarly the DeliveryCount property and the deliveryCount parameter are the same.
            _logger.LogInformation("Delivery Count: {count}", message.DeliveryCount);
            _logger.LogInformation("Delivery Count: {count}", deliveryCount);
        }
        //<docsnippet_servicebus_message_actions>
        [Function(nameof(ServiceBusMessageActionsFunction))]
        public async Task ServiceBusMessageActionsFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection", AutoCompleteMessages = false)]
            ServiceBusReceivedMessage message,
            ServiceBusMessageActions messageActions)
        {
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message Body: {body}", message.Body);
            _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);

            // Complete the message
            await messageActions.CompleteMessageAsync(message);
        }
        //</docsnippet_servicebus_message_actions>
    }
}

此示例演示一个 C# 函数,该函数接收多个服务总线队列消息,将其写入日志,然后将消息处置为已完成:

// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using System.Threading.Tasks;
using Azure.Messaging.ServiceBus;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;

namespace SampleApp
{
    /// <summary>
    /// Samples demonstrating binding to the <see cref="ServiceBusReceivedMessage"/> type.
    /// </summary>
    public class ServiceBusReceivedMessageFunctions
    {
        //<docsnippet_servicebusmessage_createlogger>
        private readonly ILogger<ServiceBusReceivedMessageFunctions> _logger;

        public ServiceBusReceivedMessageFunctions(ILogger<ServiceBusReceivedMessageFunctions> logger)
        {
            _logger = logger;
        }
        //</docsnippet_servicebusmessage_createlogger>
        /// <summary>
        /// This function demonstrates binding to a single <see cref="ServiceBusReceivedMessage"/>.
        /// </summary>
        //<docsnippet_servicebus_readmessage>
        [Function(nameof(ServiceBusReceivedMessageFunction))]
        [ServiceBusOutput("outputQueue", Connection = "ServiceBusConnection")]
        public string ServiceBusReceivedMessageFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection")] ServiceBusReceivedMessage message)
        {
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message Body: {body}", message.Body);
            _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);

            var outputMessage = $"Output message created at {DateTime.Now}";
            return outputMessage;
        }
        //</docsnippet_servicebus_readmessage>
        /// <summary>
        /// This function demonstrates binding to an array of <see cref="ServiceBusReceivedMessage"/>.
        /// Note that when doing so, you must also set the <see cref="ServiceBusTriggerAttribute.IsBatched"/> property
        /// to <value>true</value>.
        /// </summary>
        //<docsnippet_servicebus_readbatch>
        [Function(nameof(ServiceBusReceivedMessageBatchFunction))]
        public void ServiceBusReceivedMessageBatchFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection", IsBatched = true)] ServiceBusReceivedMessage[] messages)
        {
            foreach (ServiceBusReceivedMessage message in messages)
            {
                _logger.LogInformation("Message ID: {id}", message.MessageId);
                _logger.LogInformation("Message Body: {body}", message.Body);
                _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);
            }
        }
        //</docsnippet_servicebus_readbatch>
        /// <summary>
        /// This functions demonstrates that it is possible to bind to both the ServiceBusReceivedMessage and any of the supported binding contract
        /// properties at the same time. If attempting this, the ServiceBusReceivedMessage must be the first parameter. There is not
        /// much benefit to doing this as all of the binding contract properties are available as properties on the ServiceBusReceivedMessage.
        /// </summary>
        [Function(nameof(ServiceBusReceivedMessageWithStringProperties))]
        public void ServiceBusReceivedMessageWithStringProperties(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection")]
            ServiceBusReceivedMessage message, string messageId, int deliveryCount)
        {
            // The MessageId property and the messageId parameter are the same.
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message ID: {id}", messageId);

            // Similarly the DeliveryCount property and the deliveryCount parameter are the same.
            _logger.LogInformation("Delivery Count: {count}", message.DeliveryCount);
            _logger.LogInformation("Delivery Count: {count}", deliveryCount);
        }
        //<docsnippet_servicebus_message_actions>
        [Function(nameof(ServiceBusMessageActionsFunction))]
        public async Task ServiceBusMessageActionsFunction(
            [ServiceBusTrigger("queue", Connection = "ServiceBusConnection", AutoCompleteMessages = false)]
            ServiceBusReceivedMessage message,
            ServiceBusMessageActions messageActions)
        {
            _logger.LogInformation("Message ID: {id}", message.MessageId);
            _logger.LogInformation("Message Body: {body}", message.Body);
            _logger.LogInformation("Message Content-Type: {contentType}", message.ContentType);

            // Complete the message
            await messageActions.CompleteMessageAsync(message);
        }
        //</docsnippet_servicebus_message_actions>
    }
}

以下 Java 函数使用 @ServiceBusQueueTrigger中的 @ServiceBusQueueTrigger 注释来说明服务总线队列触发器的配置。 此函数获取放置在队列上的消息,然后将其添加到日志。

@FunctionName("sbprocessor")
 public void serviceBusProcess(
    @ServiceBusQueueTrigger(name = "msg",
                             queueName = "myqueuename",
                             connection = "myconnvarname") String message,
   final ExecutionContext context
 ) {
     context.getLogger().info(message);
 }

将消息添加到服务总线主题时,也可触发 Java 函数。 以下示例使用 @ServiceBusTopicTrigger 注释来说明触发器配置。

@FunctionName("sbtopicprocessor")
    public void run(
        @ServiceBusTopicTrigger(
            name = "message",
            topicName = "mytopicname",
            subscriptionName = "mysubscription",
            connection = "ServiceBusConnection"
        ) String message,
        final ExecutionContext context
    ) {
        context.getLogger().info(message);
    }

此示例使用从ServiceBusReceivedMessage获得的 ServiceBusMessageContext SDK 类型,由服务总线触发器提供:

import '@azure/functions-extensions-servicebus'; // Ensure the Service Bus extension is imported
import { app, InvocationContext } from '@azure/functions';
import { ServiceBusMessageContext } from '@azure/functions-extensions-servicebus';

//This a SDKbinding = true
export async function serviceBusQueueTrigger(
    serviceBusMessageContext: ServiceBusMessageContext,
    context: InvocationContext
): Promise<void> {
    const message = serviceBusMessageContext.messages[0];
    context.log(message);
    
    // Get current retry count from custom properties, default to 0
    const currentRetryCount = message.applicationProperties?.retryCnt ? parseInt(message.applicationProperties.retryCnt as string) : 0;
    context.log(`Current retry count: ${currentRetryCount}`);
        
        if (currentRetryCount >= 3) {
            // After 3 retries, complete the message to remove it from the queue
            context.log(`Maximum retry count (3) reached. Completing message to prevent infinite loop.`);
            await serviceBusMessageContext.actions.complete(message);
            context.log('Message completed after maximum retries');
        } else {
            // Abandon with updated retry count
            const newRetryCount = currentRetryCount + 1;
            const propertiesToModify = {
                retryCnt: newRetryCount.toString(),
                lastRetryTime: new Date().toISOString(),
                errorMessage: "Processing failed"
            };
            
            context.log(`Abandoning message with retry count: ${newRetryCount}`);
            await serviceBusMessageContext.actions.abandon(message, propertiesToModify);
        }
    
    
    context.log('triggerMetadata: ', context.triggerMetadata);
    context.log('Message body:', message.body);
}

app.serviceBusQueue('serviceBusQueueTrigger1', {
    connection: 'ServiceBusConnection',
    queueName: 'testqueue',
    sdkBinding: true,
    autoCompleteMessages: false,
    cardinality: 'many',
    handler: serviceBusQueueTrigger,
});

有关使用 SDK 类型的另一个示例,请参阅 指数退避策略示例

有关详细信息,请参阅 Node.js 参考文章中的 SDK 类型

以下示例显示了服务总线触发器 TypeScript 函数。 此函数将读取消息元数据并记录服务总线队列消息。

import { app, InvocationContext } from '@azure/functions';

export async function serviceBusQueueTrigger1(message: unknown, context: InvocationContext): Promise<void> {
    context.log('Service bus queue function processed message:', message);
    context.log('EnqueuedTimeUtc =', context.triggerMetadata.enqueuedTimeUtc);
    context.log('DeliveryCount =', context.triggerMetadata.deliveryCount);
    context.log('MessageId =', context.triggerMetadata.messageId);
}

app.serviceBusQueue('serviceBusQueueTrigger1', {
    connection: 'MyServiceBusConnection',
    queueName: 'testqueue',
    handler: serviceBusQueueTrigger1,
});

以下示例显示了服务总线触发器 JavaScript 函数。 此函数将读取消息元数据并记录服务总线队列消息。

const { app } = require('@azure/functions');

app.serviceBusQueue('serviceBusQueueTrigger1', {
    connection: 'MyServiceBusConnection',
    queueName: 'testqueue',
    handler: (message, context) => {
        context.log('Service bus queue function processed message:', message);
        context.log('EnqueuedTimeUtc =', context.triggerMetadata.enqueuedTimeUtc);
        context.log('DeliveryCount =', context.triggerMetadata.deliveryCount);
        context.log('MessageId =', context.triggerMetadata.messageId);
    },
});

以下示例演示了 function.json 文件中的服务总线触发器绑定以及使用该绑定的 PowerShell 函数

下面是 function.json 文件中的绑定数据:

{
  "bindings": [
    {
      "name": "mySbMsg",
      "type": "serviceBusTrigger",
      "direction": "in",
      "topicName": "mytopic",
      "subscriptionName": "mysubscription",
      "connection": "AzureServiceBusConnectionString"
    }
  ]
}

下面是发送服务总线消息时运行的函数。

param([string] $mySbMsg, $TriggerMetadata)

Write-Host "PowerShell ServiceBus queue trigger function processed message: $mySbMsg"

此示例使用 SDK 类型直接访问服务总线触发器提供的基础 ServiceBusReceivedMessage 对象:

import logging

import azure.functions as func
import azurefunctions.extensions.bindings.servicebus as servicebus

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

@app.service_bus_queue_trigger(arg_name="receivedmessage",
                               queue_name="QUEUE_NAME",
                               connection="SERVICEBUS_CONNECTION")
def servicebus_queue_trigger(receivedmessage: servicebus.ServiceBusReceivedMessage):
    logging.info("Python ServiceBus queue trigger processed message.")
    logging.info("Receiving: %s\n"
                 "Body: %s\n"
                 "Enqueued time: %s\n"
                 "Lock Token: %s\n"
                 "Message ID: %s\n"
                 "Sequence number: %s\n",
                 receivedmessage,
                 receivedmessage.body,
                 receivedmessage.enqueued_time_utc,
                 receivedmessage.lock_token,
                 receivedmessage.message_id,
                 receivedmessage.sequence_number)

该函数读取类型的各种属性 ServiceBusReceivedMessage 并记录它们。

有关使用服务总线 SDK 类型的更多示例,请参阅 ServiceBusReceivedMessage 示例。 有关如何在函数应用中包括 SDK 类型绑定的分步教程,请遵循 适用于服务总线的 Python SDK 绑定示例

注意

已知限制包括:

  • 不支持该 message 属性。
  • 批处理消息支持需要 4.1039 或更高版本的 Functions 运行时。

若要了解详细信息,包括支持其他 SDK 类型绑定的内容,请参阅 SDK 类型绑定

此示例演示如何通过触发器读取服务总线队列消息。 该示例取决于使用的是 v1 还是 v2 Python 编程模型

import logging
import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="ServiceBusQueueTrigger1")
@app.service_bus_queue_trigger(arg_name="msg", 
                               queue_name="<QUEUE_NAME>", 
                               connection="<CONNECTION_SETTING>")
def test_function(msg: func.ServiceBusMessage):
    logging.info('Python ServiceBus queue trigger processed message: %s',
                 msg.get_body().decode('utf-8'))

下面的示例演示了如何通过触发器读取服务总线队列主题。

import logging
import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="ServiceBusTopicTrigger1")
@app.service_bus_topic_trigger(arg_name="message", 
                               topic_name="TOPIC_NAME", 
                               connection="CONNECTION_SETTING", 
                               subscription_name="SUBSCRIPTION_NAME")
def test_function(message: func.ServiceBusMessage):
    message_body = message.get_body().decode("utf-8")
    logging.info("Python ServiceBus topic trigger processed message.")
    logging.info("Message Body: " + message_body)

特性

进程内独立工作进程 C# 库都使用 ServiceBusTriggerAttribute 特性来定义函数触发器。 C# 脚本改用 function.json 配置文件,如 C# 脚本指南中所述。

下表说明了可使用此触发器特性设置的属性:

属性 说明
队列名称 要监视的队列的名称。 仅在监视队列的情况下设置,不为主题设置。
TopicName 要监视的主题的名称。 仅在监视主题的情况下设置,不为队列设置。
SubscriptionName 要监视的订阅的名称。 仅在监视主题的情况下设置,不为队列设置。
连接 指定如何连接到服务总线的应用设置或设置集合的名称。 请参阅连接
IsBatched 消息分批传送。 需要数组或集合类型。
IsSessionsEnabled 如果连接到true队列或订阅,则为 。 否则为 false(默认值)。
AutoCompleteMessages 如果触发器应在成功调用后自动完成消息,则为 true。 如果不应,例如false时,则为 。 如果未显式设置,则行为基于autoCompleteMessages中的host.json配置

在本地开发时,需要将应用程序设置添加到 集合中的 Values中。

修饰符

仅适用于 Python v2 编程模型。

对于使用修饰器定义的 Python v2 功能,支持 service_bus_queue_trigger 上的以下属性:

属性 说明
arg_name 变量的名称,表示函数代码中的队列或主题消息。
queue_name 要监视的队列的名称。 仅在监视队列的情况下设置,不为主题设置。
connection 指定如何连接到服务总线的应用设置或设置集合的名称。 请参阅连接

对于使用 function.json 定义的 Python 函数,请参阅“配置”部分。

批注

使用 ServiceBusQueueTrigger 注释可以创建在创建服务总线队列消息时要运行的函数。 可用的配置选项包括以下属性:

属性 说明
名字 变量的名称,表示函数代码中的队列或主题消息。
队列名称 要监视的队列的名称。 仅在监视队列的情况下设置,不为主题设置。
topicName 要监视的主题的名称。 仅在监视主题的情况下设置,不为队列设置。
订阅名称 要监视的订阅的名称。 仅在监视主题的情况下设置,不为队列设置。
连接 指定如何连接到服务总线的应用设置或设置集合的名称。 请参阅连接

使用 ServiceBusTopicTrigger 注释可以指定主题和订阅,以便以触发函数的数据为目标。

在本地开发时,需要将应用程序设置添加到 集合中的 Values中。

有关更多详细信息,请参阅触发器示例

配置

仅适用于 Python v1 编程模型

下表说明了可以在传递给方法“options”或“app.serviceBusQueue()”的对象“app.serviceBusTopic()”上设置的属性。

属性 说明
队列名称 要监视的队列的名称。 仅在监视队列的情况下设置,不为主题设置。
topicName 要监视的主题的名称。 仅在监视主题的情况下设置,不为队列设置。
订阅名称 要监视的订阅的名称。 仅在监视主题的情况下设置,不为队列设置。
连接 指定如何连接到服务总线的应用设置或设置集合的名称。 请参阅连接
accessRights 连接字符串的访问权限。 可用值为 managelisten。 默认值是 manage,其指示 connection 具有“管理”权限。 如果使用不具有“管理”权限的连接字符串,请将 accessRights 设置为“listen”。 否则,Functions 运行时可能会在尝试执行需要管理权限的操作时失败。 在 Azure Functions 版本 2.x 及更高版本中,此属性不可用,因为最新版本的服务总线 SDK 不支持管理操作。
isSessionsEnabled 如果连接到true队列或订阅,则为 。 否则为 false(默认值)。
autoComplete 对于非 C# 函数,此项必须为 true,这意味着触发器应在处理后自动调用 complete,或者你会通过函数代码手动调用 complete。

设置为 true 时,触发器会在函数执行成功完成的情况下自动完成该消息,否则会放弃该消息。

函数中的异常会导致后台的运行时调用 abandonAsync 。 如果未发生异常,则在后台调用 completeAsync。 此属性仅在 Azure Functions 2.x 和更高版本中可用。

在本地开发时,需要将应用程序设置添加到 集合中的 Values中。

下表解释了在 function.json 文件中设置的绑定配置属性。

“function.json”属性 说明
类型 必须设置为 serviceBusTrigger。 在 Azure 门户中创建触发器时,会自动设置此属性。
方向 必须设置为“in”。 在 Azure 门户中创建触发器时,会自动设置此属性。
名字 变量的名称,表示函数代码中的队列或主题消息。
队列名称 要监视的队列的名称。 仅在监视队列的情况下设置,不为主题设置。
topicName 要监视的主题的名称。 仅在监视主题的情况下设置,不为队列设置。
订阅名称 要监视的订阅的名称。 仅在监视主题的情况下设置,不为队列设置。
连接 指定如何连接到服务总线的应用设置或设置集合的名称。 请参阅连接
accessRights 连接字符串的访问权限。 可用值为 managelisten。 默认值是 manage,其指示 connection 具有“管理”权限。 如果使用不具有“管理”权限的连接字符串,请将 accessRights 设置为“listen”。 否则,Functions 运行时可能会在尝试执行需要管理权限的操作时失败。 在 Azure Functions 版本 2.x 及更高版本中,此属性不可用,因为最新版本的服务总线 SDK 不支持管理操作。
isSessionsEnabled 如果连接到true队列或订阅,则为 。 否则为 false(默认值)。
autoComplete 对于非 C# 函数,此项必须为 true,这意味着触发器应在处理后自动调用 complete,或者你会通过函数代码手动调用 complete。

设置为 true 时,触发器会在函数执行成功完成的情况下自动完成该消息,否则会放弃该消息。

函数中的异常会导致后台的运行时调用 abandonAsync 。 如果未发生异常,则在后台调用 completeAsync。 此属性仅在 Azure Functions 2.x 和更高版本中可用。

在本地开发时,需要将应用程序设置添加到 集合中的 Values中。

有关完整示例,请参阅示例部分

使用情况

所有 C# 形式和扩展版本都支持以下参数类型:

类型 说明
System.String 当消息为简单文本时使用。
byte[] 用于二进制数据消息。
对象 当消息包含 JSON 时,Functions 会尝试将 JSON 数据反序列化为已知的普通旧 CLR 对象类型。

特定于消息的参数类型包含其他消息元数据。 服务总线触发器支持的特定类型取决于 Functions 运行时版本、扩展包版本和所使用的 C# 形式。

如果希望函数处理单条消息,服务总线触发器可以绑定到以下类型:

类型 说明
string 字符串格式的消息。 当消息为简单文本时使用。
byte[] 消息的字节数。
JSON 可序列化类型 当事件包含 JSON 数据时,Functions 会尝试将 JSON 数据反序列化为普通的旧 CLR 对象 (POCO) 类型。
ServiceBusReceivedMessage1 消息对象。

绑定到ServiceBusReceivedMessage时,还可以选择包含 ServiceBusMessageActions1,2 类型的参数来执行消息解决作。

如果希望函数处理一批消息,服务总线触发器可以绑定到以下类型:

类型 说明
T[],其中 T 是单消息类型之一 批处理中的事件数组。 每个条目表示一个事件。

绑定到ServiceBusReceivedMessage[]时,还可以选择包含 ServiceBusMessageActions1,2 类型的参数来执行消息解决作。

1 要使用这些类型,需要引用 Microsoft.Azure.Functions.Worker.Extensions.ServiceBus 5.14.1 或更高版本以及 SDK 类型绑定的常见依赖项

2 使用时,将ServiceBusMessageActions的属性设置为 AutoCompleteMessages。 这可以防止运行时在成功调用函数后尝试完成消息。

未定义 Connection 属性时,Functions 会查找名为 AzureWebJobsServiceBus 的应用设置,这是服务总线连接字符串的默认名称。 还可以设置 Connection 属性来指定应用程序设置(其中包含要使用的服务总线连接字符串)的名称。

可以通过 ServiceBusQueueMessageServiceBusTopicMessage 参数获取传入的服务总线消息。

将队列或主题消息作为函数的第一个参数访问。 服务总线消息作为字符串或 JSON 对象传递到函数中。

可以通过在 function.json 文件的名称属性中配置的参数来使用服务总线实例。

队列消息可通过类型为 func.ServiceBusMessage 的参数提供给函数。 服务总线消息作为字符串或 JSON 对象传递到函数中。

Functions 还支持适用于 Azure 服务总线的 Python SDK 类型绑定,它使你能够使用这些基础 SDK 类型处理数据:

重要

支持 Python 中的服务总线 SDK 类型支持为预览版,并且仅支持 Python v2 编程模型。 有关详细信息,请参阅 Python 中的 SDK 类型

有关完整示例,请参阅示例部分

连接

connection属性是应用设置中一个键的引用,返回函数运行时用于连接扩展所用服务总线实例的值。 连接属性设置的值取决于连接类型:

  • 管理身份连接:该connection属性是由<CONNECTION_NAME_PREFIX>一组设置共享的,这些设置共同定义了基于身份的连接与服务总线的连接。 更多信息请参见 定义身份连接
  • 密钥保管库 引用connection属性设置返回一个 Azure 密钥保管库 引用,指向该 连接字符串 中心维护的位置。 更多信息请参见定义 密钥保管库 连接。
  • App Configuration 引用connection属性设置返回一个 Azure 应用程序配置 引用,返回一个 连接字符串 或 密钥保管库 引用。 更多信息请参见连接文章中的 Azure 应用程序配置
  • Connection stringconnection属性设置返回 服务总线 实例的实际 连接字符串。 由于连接字符串包含共享的秘密密钥,你应尽量考虑使用管理身份连接。 更多信息请参见定义连接。

欲了解更多关于绑定连接的信息,请参见 Azure Functions 中的 Manage connection

若要获取连接字符串,请执行获取管理凭据中显示的步骤。 必须是服务总线命名空间的连接字符串,不限于特定的队列或主题。

如果应用设置名称以 AzureWebJobs开头,你可以只指定名称的其余部分。 例如,如果设置为connectionMyServiceBus,Functions 运行时将查找名为AzureWebJobsMyServiceBus“的应用”设置。 如果留connection空,函数运行时会使用应用设置中名为 AzureWebJobsServiceBus的默认服务总线 连接字符串。

权限缩放

服务总线扩展使用服务总线 Administration API(GetQueueRuntimePropertiesAsync / GetSubscriptionRuntimePropertiesAsync)来获取准确的消息计数,用于规模决策。 该 API 需要超出发送或接收消息所需的权限:

  • SAS 连接字符串:SAS 策略必须包含 管理 访问权。
  • 基于身份的连接:身份必须被分配为 Azure 服务总线 数据所有者角色,或包含 的自定义角色Microsoft.ServiceBus/namespaces/*/read

当连接缺少这些权限时,启动时不会看到错误。 相反,扩展默默地回归到基于窥视的消息估计,这种方法的准确性较低,可能导致延迟或错误的扩展决策。

Tip

对于依赖自动扩展的生产工作负载,可以包含管理访问权(SAS)或分配 Azure 服务总线 数据所有者角色(基于身份的连接),以确保规模行为的准确性。 应用程序设置中的连接字符串,名为 AzureWebJobsServiceBus

有害消息

无法在 Azure Functions 中控制或配置有害消息处理。 服务总线处理有害消息本身。

PeekLock 行为

Functions 运行时以 PeekLock 模式接收消息。

默认情况下,如果函数成功完成,则运行时会对此消息调用 Complete;如果函数失败,则调用 Abandon。 可以通过 autoCompleteMessages 中的 host.json 属性禁用自动完成。

默认情况下,如果函数成功完成,则运行时会对此消息调用 Complete;如果函数失败,则调用 Abandon。 可以通过 autoCompleteMessages 中的 host.json 属性或通过触发器特性上的一个属性来禁用自动完成。 如果你的函数代码处理消息结算,则应禁用自动完成。

如果函数的运行时间长于 PeekLock 超时时间,则只要该函数正在运行,就会自动续订锁定。 maxAutoRenewDurationhost.json 中可配置,它映射到 serviceBusProcessor.MaxAutoLockRenewalDuration。 此设置的默认值为 5 分钟。

消息元数据

使用特定于消息传送的类型,你可以轻松地检索作为对象属性的元数据。 这些属性取决于 Functions 运行时版本、扩展包版本和所使用的 C# 形式。

这些属性是 ServiceBusReceivedMessage 类的成员。

属性 类型 说明
ApplicationProperties ApplicationProperties 由发送方设置的属性。
ContentType string 发送方和接收方用于实现应用程序特定逻辑的内容类型标识符。
CorrelationId string 相关 ID。
DeliveryCount Int32 传递次数。
EnqueuedTime DateTime 排队时间 (UTC)。
ScheduledEnqueueTimeUtc DateTime 计划的排队时间 (UTC)。
ExpiresAt DateTime 到期时间 (UTC)。
MessageId string 服务总线可用于标识重复消息的用户定义值(如果启用)。
ReplyTo string 对队列地址的回复。
Subject string 特定于应用程序的标签,可用于替代 Label 元数据属性。
To string 发送到地址。

后续步骤