了解 cosmos db 中的 changefeed

Question

下面是我的代码，我试图将某些文档从一个容器复制到另一个容器。我将 Azure 函数与带有更改源的 Cosmos db 触发器一起使用。当我运行它处于调试模式时，我没有得到任何文档，并且在控制台中它显示为主机锁定状态，这意味着没有新文档可读。根据我的理解，azure 函数应该从我的第一个运行开始读取，然后在随后的运行中它应该从它离开的地方开始读取租用容器中的令牌。正在寻找专家来确认这一点，我没有在这里阅读任何项目的原因是什么？我将调试器点放在第一行，但它没有命中它。

[FunctionName( "MigrateWithChangeFeed" )]
        public static async Task Run( [CosmosDBTrigger(
            databaseName: "sourcedb",
            collectionName: "collec1",
            StartFromBeginning = true,
            ConnectionStringSetting = "CosmosDBConnection",
            LeaseCollectionName = "leases",            
            CreateLeaseCollectionIfNotExists = true
    )] IReadOnlyList<Document> source,
    ILogger log )
        {           
            Uri taskCollectionUri = UriFactory.CreateDocumentCollectionUri( source_databaseId, source_containerId );
            DocumentClient client = new DocumentClient( new Uri( _endpointUrl ), _primaryKey );
            if( source != null && source.Count > 0 )
            {
                foreach( var item in source )
                {
                    string customerName = string.Empty;
                    var jsonContent = (JObject)JsonConvert.DeserializeObject( item.ToString() );
                    if( !string.IsNullOrEmpty( (string)jsonContent["ClassId"] ) )
                    {
                        customerName = (string)jsonContent["Customer"];
                        if( string.IsNullOrEmpty( customerName ) )
                        {
                            customerName = projectCustomerNameMapping
                                .FirstOrDefault( q => q.Key.IndexOf( (string)jsonContent["Project"] ) != -1 ||
                                ((string)jsonContent["Project"]).IndexOf( q.Key ) != -1 ).Value ?? "Unknown";
                        }
                        string partitionkeyValue = string.Concat( (string)jsonContent["ClassId"],
                            "|",
                            (string)jsonContent["Project"],
                            "|",
                            customerName );
                        jsonContent.Add( new JProperty( "PartitionKey", partitionkeyValue ) );
                        await client.CreateDocumentAsync( UriFactory.CreateDocumentCollectionUri(
                       source_databaseId, target_containerId ),
                       jsonContent );
                    }

                }
            }

Answer 1

首先，请不要在函数内部创建 DocumentClient，您应该使用 singleton/static 实例，否则您将很快运行退出可用连接。参考：https://docs.microsoft.com/azure/azure-functions/manage-connections#static-clients

Change Feed 的工作原理是，您第一次运行此函数时，它将从头读取（基本上读取受监视集合中的所有现有文档），之后（假设它仍然运行ning)，它将拾取任何新创建或更新的文档。 Trigger主要检查包含标记先前存储的检查点的文档的租约集合，如果有none，则表示它是第一次启动，因此它尊重StartFromBeginning标志，之后，当有文档时，它会忽略它。

不过请记住，如果您在本地运行启用此功能并且还在云中部署，那么它们将充当同一功能的多个实例 (https://docs.microsoft.com/azure/cosmos-db/troubleshoot-changefeed-functions#some-changes-are-missing-in-my-trigger) unless each one is using a different lease collection or using a lease prefix (https://docs.microsoft.com/azure/cosmos-db/how-to-create-multiple-cosmos-db-triggers) .

了解 cosmos db 中的 changefeed

understanding on changefeed in cosmos db

azure-functions

azure-cosmosdb