部署时多次调用 Azure 服务总线队列触发器函数

Question

我有两个 Azure Functions。一个是HTTP触发的，我们称之为API，另一个是ServiceBusQueue触发，我们称之为Listener。

第一个（API）将 HTTP 请求放入队列，第二个（监听器）拾取并处理该请求。函数SDK版本为：3.0.7。

为此，我的解决方案中有两个项目。一个包含 Azure Functions，另一个包含服务。 API 一旦被触发，就会调用另一个项目的服务，将消息放入队列。而Listener一旦收到消息，就会从服务项目中调用一个服务来处理消息。

任何长时间运行的进程？

Listener 实际上执行了一个轻量级的工作流程，考虑到它执行的工作量，这一切都发生得非常快。平均执行时间为90 seconds。

队列规范是什么？

侦听器侦听并托管在 Azure ServiceBus 命名空间中的队列具有以下属性集：

• 最大交付次数：1
• 消息生存时间：1 天
• 自动删除：从不
• 重复检测窗口：10 分钟
• 消息锁定持续时间：5 分钟

这里是它的截图：

API 使用以下方法将 HTTP 请求放入队列：

 public async Task ProduceAsync(string queueName, string jsonMessage)
 {
        jsonMessage.NotNull();
        queueName.NotNull();

        IQueueClient client = new QueueClient(Environment.GetEnvironmentVariable("ServiceBusConnectionString"), queueName, ReceiveMode.PeekLock)
        {
            OperationTimeout = TimeSpan.FromMinutes(5)
        };

        await client.SendAsync(new Message(Encoding.UTF8.GetBytes(jsonMessage)));

        if (!client.IsClosedOrClosing)
        {
            await client.CloseAsync();
        }
 }

而Listener（服务总线队列触发azure函数），有如下代码处理消息：

[FunctionName(nameof(UpdateBookingCalendarListenerFunction))]
public async Task Run([ServiceBusTrigger(ServiceBusConstants.UpdateBookingQueue, Connection = ServiceBusConstants.ConnectionStringKey)] string message)
{
        var data = JsonConvert.DeserializeObject<UpdateBookingCalendarRequest>(message);
        _telemetryClient.TrackTrace($"{nameof(UpdateBookingCalendarListenerFunction)} picked up a message at {DateTime.Now}. Data: {data}");

        await _workflowHandler.HandleAsync(data);
}

问题

Listener 函数处理相同的消息3 次！我不知道为什么！我在 Google 上搜索并阅读了一些 StackOverFlow 线程，例如 this one。看起来每个人都建议确保lock duration 足够长，以使该过程完全执行。虽然，我已经为锁输入了5 minutes，但是问题仍然存在。我非常感谢您对此的任何帮助。

Answer 1

只需在此处添加此内容，可能对其他人有所帮助。

经过更多调查后，我意识到在我的特定情况下，问题与 Azure Functions 和服务总线无关。在UpdateBookingCalendarListenerFunction 向其发送消息的工作流处理程序中，我试图以并行方式调用一些外部API，但是，由于某些未知原因（对我而言），处理程序代码再次调用外部API，不管它迭代了多少条记录。下面的代码展示了我是如何实现并行 API 调用的，其他代码展示了我是如何一一完成的，最终解决了我遇到的问题。

我的原始代码 - 并行调用 API

 public async Task<IEnumerable<StaffMemberGraphApiResponse>> AddAdminsAsync(IEnumerable<UpdateStaffMember> admins, string bookingId)
    {
        var apiResults = new List<StaffMemberGraphApiResponse>();

        var adminsToAdd = admins.Where(ad => ad.Action == "add");

        _telemetryClient.TrackTrace($"{nameof(UpdateBookingCalendarWorkflowDetailHandler)} Recognized {adminsToAdd.Count()} admins to add to booking with id: {bookingId}");

        var addAdminsTasks = adminsToAdd.Select(admin => _addStaffGraphApiHandler.HandleAsync(new AddStaffToBookingGraphApiRequest
        {
            BookingId = bookingId,
            DisplayName = admin.DisplayName,
            EmailAddress = admin.EmailAddress,
            Role = StaffMemberAllowedRoles.Admin
        }));

        if (addAdminsTasks.Any())
        {
            var addAdminsTasksResults = await Task.WhenAll(addAdminsTasks);
            apiResults = _populateUpdateStaffMemberResponse.Populate(addAdminsTasksResults, StaffMemberAllowedRoles.Admin).ToList();
        }

        return apiResults;
    }

我的新代码没有将 API 调用聚合到 addAdminsTasks 对象中，因此没有 await Task.WhenAll(addAdminsTasks)：

 public async Task<IEnumerable<StaffMemberGraphApiResponse>> AddStaffMembersAsync(IEnumerable<UpdateStaffMember> members, string bookingId, string targetRole)
    {
        var apiResults = new List<StaffMemberGraphApiResponse>();

        foreach (var item in members.Where(v => v.Action == "add"))
        {
            _telemetryClient.TrackTrace($"{nameof(UpdateBookingCalendarWorkflowDetailHandler)} Adding {targetRole} to booking: {bookingId}. data: {JsonConvert.SerializeObject(item)}");

            apiResults.Add(_populateUpdateStaffMemberResponse.PopulateAsSingleItem(await _addStaffGraphApiHandler.HandleAsync(new AddStaffToBookingGraphApiRequest
            {
                BookingId = bookingId,
                DisplayName = item.DisplayName,
                EmailAddress = item.EmailAddress,
                Role = targetRole
            }), targetRole));
        }

        return apiResults;
    }

我研究了第一种方法，任务数量与IEnumerable 输入的数量完全匹配，但是，API 又被调用了一次。在_addStaffGraphApiHandler.HandleAsync 中，实际上只有一个引发POSTrequest 的HttpClient 对象。无论如何，使用第二个代码已经解决了这个问题。