如何在Azure中通过Service Bus消息动态创建并销毁ACI容器

整体流程 #

• Service Bus主题接收客户端发送的导出请求消息（包含任务参数：数据源、目标存储路径等）
• Azure Function（Service Bus触发器）监听主题订阅，收到消息后触发ACI容器创建
• ACI容器拉取指定Docker镜像，执行导出任务
• 任务完成后，容器主动触发销毁逻辑（或由外部服务检测状态后销毁）
• 删除ACI容器资源，释放计算资源

核心实现步骤 #

1. 前置配置 #

• 创建Service Bus主题和订阅，确保订阅设置PeekLock消息接收模式
• 准备好处理导出任务的Docker镜像，推送到Azure Container Registry（ACR）或公开镜像仓库
• 为Azure Function分配Contributor权限（或更细粒度的ACI创建/删除权限），使其能操作Azure容器实例资源

2. 监听Service Bus消息并创建ACI容器 #

使用Azure Function + Azure .NET SDK实现消息监听与ACI创建逻辑：

using Azure.Messaging.ServiceBus;
using Azure.ResourceManager.Containers;
using Azure.ResourceManager.Containers.Models;
using Azure.ResourceManager;
using System.Text.Json;
using System.Threading.Tasks;

public class ExportRequestProcessor
{
    private readonly ArmClient _armClient;
    private readonly string _resourceGroupName = "your-resource-group";
    private readonly string _acrLoginServer = "your-acr.azurecr.io";
    private readonly string _imageName = "export-task-image:latest";
    private readonly string _subscriptionId = "your-subscription-id";

    public ExportRequestProcessor(ArmClient armClient)
    {
        _armClient = armClient;
    }

    public async Task ProcessMessageAsync(ServiceBusReceivedMessage message, ServiceBusMessageActions messageActions)
    {
        try
        {
            // 解析消息中的导出任务参数
            var exportParams = JsonSerializer.Deserialize<ExportTaskParameters>(message.Body.ToString());

            // 构建ACI容器组配置
            var containerGroupData = new ContainerGroupData(AzureLocation.EastUs)
            {
                Containers =
                {
                    new ContainerInstance($"export-task-container-{Guid.NewGuid()}")
                    {
                        Image = $"{_acrLoginServer}/{_imageName}",
                        Resources = new ContainerResourceRequirements(new ContainerResourceRequests(cpu: 1, memoryInGB: 2)),
                        EnvironmentVariables =
                        {
                            new EnvironmentVariable("DATA_SOURCE", exportParams.DataSource),
                            new EnvironmentVariable("TARGET_PATH", exportParams.TargetPath),
                            new EnvironmentVariable("RESOURCE_GROUP_NAME", _resourceGroupName),
                            new EnvironmentVariable("SUBSCRIPTION_ID", _subscriptionId)
                        }
                    }
                },
                OsType = ContainerGroupOsType.Linux,
                ImageRegistryCredentials =
                {
                    new ImageRegistryCredential(_acrLoginServer)
                    {
                        Username = "acr-username",
                        Password = "acr-password"
                    }
                }
            };

            // 创建ACI容器组
            var resourceGroup = _armClient.GetResourceGroupResource(ResourceIdentifier.FromString($"/subscriptions/{_subscriptionId}/resourceGroups/{_resourceGroupName}"));
            var containerGroups = resourceGroup.GetContainerGroups();
            var containerGroupName = $"export-task-group-{Guid.NewGuid()}";
            var lro = await containerGroups.CreateOrUpdateAsync(WaitUntil.Completed, containerGroupName, containerGroupData);
            
            // 标记消息处理完成
            await messageActions.CompleteMessageAsync(message);
        }
        catch (Exception ex)
        {
            // 异常处理：将消息移至死信队列
            await messageActions.DeadLetterMessageAsync(message, errorDescription: ex.Message);
        }
    }

    public class ExportTaskParameters
    {
        public string DataSource { get; set; }
        public string TargetPath { get; set; }
    }
}

3. 任务完成后销毁ACI容器 #

推荐两种销毁方式：

• 容器主动触发：在导出任务的Docker镜像中，任务完成后调用Azure SDK删除当前ACI容器。示例容器内的Python销毁逻辑：

import os
from azure.identity import DefaultAzureCredential
from azure.mgmt.containerinstance import ContainerInstanceManagementClient

def delete_aci_container():
    subscription_id = os.environ["SUBSCRIPTION_ID"]
    resource_group = os.environ["RESOURCE_GROUP_NAME"]
    # 通过容器元数据获取容器组名称
    container_group_name = os.environ["AZURE_CONTAINER_GROUP_NAME"]

    credential = DefaultAzureCredential()
    client = ContainerInstanceManagementClient(credential, subscription_id)

    client.container_groups.delete(resource_group, container_group_name)

# 执行导出任务
def run_export_task():
    # 此处编写导出逻辑，比如读取数据源、写入目标存储
    pass

if __name__ == "__main__":
    run_export_task()
    delete_aci_container()

• 外部定时检测：用Azure Function（定时触发器）定期查询ACI容器状态，对已终止的容器执行删除操作，避免因容器异常导致资源残留。

关键注意事项 #

• 权限控制：使用Azure托管身份替代硬编码凭证，为Function分配最小必要权限（比如Microsoft.ContainerInstance/containerGroups/write和delete权限）
• 资源唯一标识：为每个ACI容器组生成唯一名称（如附加GUID），避免命名冲突
• 超时与清理：为ACI容器设置RestartPolicy = ContainerGroupRestartPolicy.Never，并配置定时清理任务处理异常终止的容器
• 日志排查：将ACI容器日志导出到Azure Log Analytics，方便追踪任务执行状态和排查问题

内容的提问来源于stack exchange，提问作者Rajib Koley