azure - 何时不对 Azure Functions ServiceBus 绑定(bind)中的 IAsyncCollector 调用 FlushAsync

Question

我有一系列 Azure Function 应用程序，它们都使用 IAsyncCollector<Message> 将消息输出到 Azure 服务总线。 :

public async Task Run([ServiceBus(...)] IAsyncCollector<Message> messages)
{
    ...
    await messages.AddAsync(msg);
}

我时不时会收到如下所示的错误记录:

Microsoft.Azure.WebJobs.Host.FunctionInvocationException: Exception while executing function: Function
 ---> Microsoft.Azure.ServiceBus.ServiceBusTimeoutException: The operation did not complete within the allocated time 00:00:59.9999536 for object message.Reference: ..., 7/13/2020 2:46:24 PM
 ---> System.TimeoutException: The operation did not complete within the allocated time 00:00:59.9999536 for object message.
   at Microsoft.Azure.Amqp.AsyncResult.End[TAsyncResult](IAsyncResult result)
   at Microsoft.Azure.Amqp.SendingAmqpLink.EndSendMessage(IAsyncResult result)
   at System.Threading.Tasks.TaskFactory`1.FromAsyncCoreLogic(IAsyncResult iar, Func`2 endFunction, Action`1 endAction, Task`1 promise, Boolean requiresSynchronization)
--- End of stack trace from previous location where exception was thrown ---
   at Microsoft.Azure.ServiceBus.Core.MessageSender.OnSendAsync(IList`1 messageList)
   --- End of inner exception stack trace ---
   at Microsoft.Azure.ServiceBus.Core.MessageSender.OnSendAsync(IList`1 messageList)
   at Microsoft.Azure.ServiceBus.RetryPolicy.RunOperation(Func`1 operation, TimeSpan operationTimeout)
   at Microsoft.Azure.ServiceBus.RetryPolicy.RunOperation(Func`1 operation, TimeSpan operationTimeout)
   at Microsoft.Azure.ServiceBus.Core.MessageSender.SendAsync(IList`1 messageList)
   at Microsoft.Azure.WebJobs.ServiceBus.Bindings.MessageSenderExtensions.SendAndCreateEntityIfNotExists(MessageSender sender, Message message, Guid functionInstanceId, EntityType entityType, CancellationToken cancellationToken)
   at My.Function.Run(String mySbMsg, IAsyncCollector`1 messages)

我很难弄清楚这种情况何时会发生。但我最近了解到FlushAsync方法:

await messages.AddAsync(msg);
await messages.FlushAsync();

我的问题如下。为什么我不包括对 FlushAsync 的调用在我的函数中？在我自己的代码中获取超时异常将使重试、更好的异常日志记录等成为可能。在函数代码中像这样手动刷新有什么缺点吗？

Answer 1

Why would I ever NOT include a call to FlushAsync in my function? Getting the timeout exception in my own code will make it possible to retry, do better exception logging, and more.

我要在这里更进一步说，在我获得了一些 Azure Functions 的经验后，我现在避免 IAsyncCollector<T>完全地。一些实现发布于 AddAsync ;其他实现可能会在 AddAsync 上发布和FlushAsync 。我怀疑服务总线实现实际上是在 AddAsync 上发布的，在这种情况下FlushAsync可能是一个空。

关于 IAsyncCollector<T> 的精彩部分是它给你一个“写这些东西”的抽象；您所要做的就是提供一个连接字符串，剩下的就是魔法了。 IAsyncCollector<T> 的问题是它给了你一个抽象，因此你的控制力要少得多。

在幕后，进行了多少次重试？他们是使用恒定的延迟还是呈指数增长？如果它永远不会成功，会出现什么行为？通常这些关键信息都不会被记录下来。

特别烦人的是 AF 团队更改抽象的语义。例如，对于某些输出绑定(bind)(CosmosDB 或存储，我不记得了)，重试行为从函数 SDK 的一个版本更改为下一个版本。

因此，我倾向于避免输出绑定(bind)，尤其是 IAsyncCollector<T> 。我通常想要进行严格但呈指数增长的去相关抖动重试，上限为一分钟左右，但当函数运行时只剩下一分钟时中止，然后恢复行为更改为将消息写入错误队列(并重试)。这比 IAsyncCollector<T> 复杂得多。可以提供，但 Polly 直接调用 SDK 并不难。

Any downsides of flushing manually like this within the function code?

没有。默认情况下，IAsyncCollector<T>.FlushAsync在函数执行后由函数主机调用。所以如果你自己调用它，你只是提前调用它。多次调用应该是安全的。