0

我有一个 azure 函数（EventGrid 触发器），当在 blob 存储（Data Lake Storage Gen2）中创建新的 blob 时触发。在这个函数中，我需要读取 blob 并将其反序列化为一个对象（类型已知）。

这是我用来获取 CloudBlob 的方法：

public async Task<CloudBlob> GetCloudBlob(Uri uri)
{
   var cloudBlockBlob = new CloudBlob(uri, _storageCredentials);
   if (await cloudBlockBlob.ExistsAsync())
       return cloudBlockBlob;
   return null;
}

使用此代码，我阅读了云 blob

using var stream = await cloudBlob.OpenReadAsync();
await _requestFileHandler.HandleFile(stream, name, prefix, cloudBlob.Uri);

在方法 HandleFile 我调用 DeserializeAsync 方法：

var model = await JsonSerializer.DeserializeAsync<RequestModel>(stream);

由于某种原因，流有时是空的（stream.Length 为 0）导致此异常。

The input does not contain any JSON tokens. Expected the input to start with a valid JSON token, when isFinalBlock is true. Path: $ | LineNumber: 0 | BytePositionInLine: 0.

有趣的是，如果我再次重试这个 blob 的过程，那么流会按预期填充。

我是否缺少任何等待条件，或者 Blob 尚未完全写入存储？

4

3 回答 3

1

对流进行序列化会使流位于末尾，所以当你去反序列化它时，没有什么可以反序列化的。

您需要添加如下一行：

stream.Position = 0;

在序列化之后和反序列化之前。

于 2020-08-28T09:32:20.530 回答
0

尝试转换await cloudBlockBlob.ExistsAsync()为await cloudBlockBlob.ExistsAsync().ConfigureAwait(false), await cloudBlob.OpenReadAsync()to await cloudBlob.OpenReadAsync().ConfigureAwait(false), await _requestFileHandler.HandleFile(stream, name, prefix, cloudBlob.Uri)toawait _requestFileHandler.HandleFile(stream, name, prefix, cloudBlob.Uri).ConfigureAwait(false)和await JsonSerializer.DeserializeAsync<RequestModel>(stream)to await JsonSerializer.DeserializeAsync<RequestModel>(stream).ConfigureAwait(false)。

当await应用关键字时，它会暂停调用方法并将控制权交还给调用者，直到等待的任务完成。我们可以避免强制延续/回调回到原始上下文，我们通过使用ConfigureAwait(false). 不使用ConfigureAwait(false)会导致性能问题和不可靠性。由于您没有使用它，因此stream有时在调用者上下文中，有时在当前上下文中。当前上下文中的一个为您提供正确的结果。您可以访问 devblogs.microsoft.com/dotnet/configureawait-faq 了解更多信息。

于 2020-08-28T20:36:59.650 回答
0

您可以为EventGridTrigger函数使用 blob 输入绑定来获取 blob 流，请参见以下代码片段：

public static async Task Run(JObject eventGridEvent, Stream body, ILogger log)
{
    log.LogInformation($"{eventGridEvent}");    
    // ...
    var model = await System.Text.Json.JsonSerializer.DeserializeAsync<RequestModel>(body);                  
    // ...
    await Task.CompletedTask;
}

和 function.json 文件：

 {
  "bindings": [
    {
      "type": "eventGridTrigger",
      "name": "eventGridEvent",
      "direction": "in"
    },
    {
      "type": "blob",
      "name": "body",
      "path": "{data.url}",
      "connection": "rk2019dlstg_STORAGE",
      "direction": "in"
    }
  ],
  "disabled": false
}
于 2020-08-29T12:39:24.970 回答