通过

你当前正在访问 Microsoft Azure Global Edition 技术文档网站。 如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

从更改馈送处理器库迁移到 Azure Cosmos DB .NET V3 SDK

  • 适用于: ✅ NoSQL

本文介绍了将使用更改源处理器库的现有应用程序代码迁移到最新版本 .NET SDK（也称为 .NET V3 SDK）中的更改源功能的必要步骤。

所需的代码更改

.NET V3 SDK 有几个重大更改，以下是迁移应用程序的关键步骤：

  1. 将 DocumentCollectionInfo 实例转换为对受监视容器和租用容器的 Container 引用。
  2. 需要将当前使用 WithProcessorOptions 的自定义项更新为使用 WithLeaseConfiguration 和 WithPollInterval 处理间隔，使用 WithStartTime处理开始时间，以及使用 WithMaxItems 来定义最大项数。
  3. 将processorName上的GetChangeFeedProcessorBuilder设置为与配置的ChangeFeedProcessorOptions.LeasePrefix的值相匹配，如果不行，则使用string.Empty。
  4. 更改不再以IReadOnlyList<Document>的形式交付，而是以IReadOnlyCollection<T>的形式，其中T是一个需要定义的类型，不再有基项类。
  5. 若要处理更改，不再需要实现 IChangeFeedObserver，而需要 定义委托。 委托可以是静态函数，或者，如果需要跨执行维护状态，则可以创建自己的类并将实例方法作为委托传递。

例如，如果构建变更流处理器的原始代码如下所示：

ChangeFeedProcessorLibrary.DocumentCollectionInfo monitoredCollectionInfo = new ChangeFeedProcessorLibrary.DocumentCollectionInfo()
{
    DatabaseName = databaseId,
    CollectionName = Program.monitoredContainer,
    Uri = new Uri(configuration["EndPointUrl"]),
    MasterKey = configuration["AuthorizationKey"]
};

ChangeFeedProcessorLibrary.DocumentCollectionInfo leaseCollectionInfo = new ChangeFeedProcessorLibrary.DocumentCollectionInfo()
{
    DatabaseName = databaseId,
    CollectionName = Program.leasesContainer,
    Uri = new Uri(configuration["EndPointUrl"]),
    MasterKey = configuration["AuthorizationKey"]
};

ChangeFeedProcessorLibrary.ChangeFeedProcessorBuilder builder = new ChangeFeedProcessorLibrary.ChangeFeedProcessorBuilder();
var oldChangeFeedProcessor = await builder
    .WithHostName("consoleHost")
    .WithProcessorOptions(new ChangeFeedProcessorLibrary.ChangeFeedProcessorOptions
    {
        StartFromBeginning = true,
        LeasePrefix = "MyLeasePrefix",
        MaxItemCount = 10,
        FeedPollDelay = TimeSpan.FromSeconds(1)
    })
    .WithFeedCollection(monitoredCollectionInfo)
    .WithLeaseCollection(leaseCollectionInfo)
    .WithObserver<ChangeFeedObserver>()
    .BuildAsync();

迁移的代码如下所示：

Container leaseContainer = client.GetContainer(databaseId, Program.leasesContainer);
Container monitoredContainer = client.GetContainer(databaseId, Program.monitoredContainer);
ChangeFeedProcessor changeFeedProcessor = monitoredContainer
    .GetChangeFeedProcessorBuilder<ToDoItem>("MyLeasePrefix", Program.HandleChangesAsync)
        .WithInstanceName("consoleHost")
        .WithLeaseContainer(leaseContainer)
        .WithMaxItems(10)
        .WithPollInterval(TimeSpan.FromSeconds(1))
        .WithStartTime(DateTime.MinValue.ToUniversalTime())
        .Build();

对于委托，你可以使用静态方法来接收事件。 如果你正在使用 IChangeFeedObserverContext 的信息，则可以迁移到 ChangeFeedProcessorContext 进行使用。

  • ChangeFeedProcessorContext.LeaseToken 可以替代 IChangeFeedObserverContext.PartitionKeyRangeId
  • ChangeFeedProcessorContext.Headers 可以用来代替 IChangeFeedObserverContext.FeedResponse
  • ChangeFeedProcessorContext.Diagnostics 包含有关请求延迟用于故障排除的详细信息
static async Task HandleChangesAsync(ChangeFeedProcessorContext context, IReadOnlyCollection<ToDoItem> changes, CancellationToken cancellationToken)
{
    Console.WriteLine($"Started handling changes for lease {context.LeaseToken}...");
    Console.WriteLine($"Change Feed request consumed {context.Headers.RequestCharge} RU.");
    // SessionToken if needed to enforce Session consistency on another client instance
    Console.WriteLine($"SessionToken ${context.Headers.Session}");

    // We may want to track any operation's Diagnostics that took longer than some threshold
    if (context.Diagnostics.GetClientElapsedTime() > TimeSpan.FromSeconds(1))
    {
        Console.WriteLine($"Change Feed request took longer than expected. Diagnostics:" + context.Diagnostics.ToString());
    }

    foreach (ToDoItem item in changes)
    {
        Console.WriteLine($"\tDetected operation for item with id {item.id}, created at {item.creationTime}.");
        // Simulate work
        await Task.Delay(1);
    }
}

系统健康事件和可观测性

如果以前你使用过 IHealthMonitor 或者正在利用 IChangeFeedObserver.OpenAsync 和 IChangeFeedObserver.CloseAsync，建议使用 通知 API。

  • IChangeFeedObserver.OpenAsync 可以替换为 WithLeaseAcquireNotification。
  • IChangeFeedObserver.CloseAsync 可以替换为 WithLeaseReleaseNotification。
  • IHealthMonitor.InspectAsync 可以替换为 WithErrorNotification。

管理和租用容器

与更改源处理器库类似，.NET V3 SDK 中的更改源功能使用 租约容器 来存储状态。 但是，架构不同。

SDK V3 更改源处理器将检测任何旧的库状态，并在首次执行迁移的应用程序代码时自动将其迁移到新架构。

可以使用旧代码安全地停止应用程序，将代码迁移到新版本，启动迁移的应用程序，以及应用程序停止时发生的任何更改，将由新版本拾取和处理。

其他资源

后续步骤

现在可以继续了解有关更改源处理器的详细信息，请参阅以下文章：

  • Last updated on