Azure 函数等待所有队列项目完成

Question

我想使用 Azure 函数对要处理的项目进行排队（使用队列和队列触发器）并等待所有队列项目完成处理，但不确定如何在隔离的工作线程中使用 .NET 8 Azure 函数项目来执行此操作。

Answer 1

这是实现此目的的高级方法和代码示例：

1。设置 Azure 队列存储和队列触发器： 创建 Azure 队列来保存需要处理的项目。 创建一个带有队列触发器的 Azure 函数来处理每条消息。

2。跟踪已处理的项目： 使用共享跟踪机制（例如数据库或缓存）来跟踪已处理的项目。处理所有项目后，您可以根据需要执行任何最终步骤。

3.实现等待逻辑： 如果您需要等待队列中的所有消息完成，您可能有一个父函数或 API 端点，它将把所有项目排入队列，然后等待它们的处理完成。这可以通过任务来完成。 WhenAll 如果处理作为任务启动。

这是一个示例实现：

第 1 步： 将项目放入 Azure 队列中 首先，将您的项目放入队列中。这可以在单独的 Azure Function 或 API 端点中完成。

public async Task EnqueueItemsAsync(IEnumerable<string> itemsToProcess)
{
    string connectionString = "<your_queue_storage_connection_string>";
    string queueName = "my-queue";

    var queueClient = new QueueClient(connectionString, queueName);
    await queueClient.CreateIfNotExistsAsync();

    foreach (var item in itemsToProcess)
    {
        await queueClient.SendMessageAsync(Convert.ToBase64String(System.Text.Encoding.UTF8.GetBytes(item)));
    }
}

第2步：使用队列触发功能处理项目 创建一个队列触发的 Azure Function 来处理每条消息。每当有新项目添加到队列时，就会自动触发此功能。

public class QueueProcessorFunction
{
    private readonly ILogger<QueueProcessorFunction> _logger;

    public QueueProcessorFunction(ILogger<QueueProcessorFunction> logger)
    {
        _logger = logger;
    }

    [Function("QueueProcessor")]
    public async Task Run([QueueTrigger("my-queue", Connection = "AzureWebJobsStorage")] string queueMessage)
    {
        // Process the item
        _logger.LogInformation($"Processing item: {queueMessage}");

        // Simulate processing delay
        await Task.Delay(1000);

        // Mark item as processed
        _logger.LogInformation($"Completed processing for item: {queueMessage}");
    }
}

第 3 步： 等待所有处理完成 如果您想等到所有消息都处理完毕，您可以实现跟踪机制，例如将处理状态保存在数据库中，然后轮询以检查是否所有项目都已完成。

或者，如果您在同一个应用程序中使用它并且需要等待代码中的处理，您可以使用内存中或分布式 Task.WhenAll 方法来等待所有处理的完成。

示例：使用 Task.WhenAll 进行队列处理 如果您在受控环境中运行排队和处理（例如使用任务进行内存中处理），则可以使用 Task.WhenAll 等待完成：

public async Task ProcessAllItemsAsync(IEnumerable<string> itemsToProcess)
{
    var tasks = new List<Task>();

    // Enqueue and process each item as a separate task
    foreach (var item in itemsToProcess)
    {
        var task = Task.Run(async () =>
        {
            await EnqueueItemAsync(item);
            // Simulate processing by waiting for the queue trigger to complete
            await ProcessQueueItemAsync(item); // Replace with real queue processing or polling logic
        });
        tasks.Add(task);
    }

    // Wait until all tasks are complete
    await Task.WhenAll(tasks);

    Console.WriteLine("All items processed.");
}

public async Task EnqueueItemAsync(string item)
{
    // Enqueue item logic here
}

public async Task ProcessQueueItemAsync(string item)
{
    // Process queue item logic here (e.g., poll for completion)
}