Azure Cosmos DB での競合解決ポリシーの管理

[アーティクル]
08/15/2024

適用対象: NoSQL

マルチリージョン書き込みを使用すると、複数のクライアントが同じ項目に書き込んだときに競合が発生する場合があります。競合が発生した場合は、さまざまな競合解決ポリシーを使用してその競合を解決できます。この記事では、競合解決ポリシーを管理する方法について説明します。

ヒント

競合解決ポリシーは、コンテナーの作成時にのみ指定できます。コンテナーの作成後には変更できません。

最終書き込み者優先競合解決ポリシーを作成する

以下のサンプルでは、最終書き込み者優先競合解決ポリシーを使用してコンテナーを設定する方法について説明します。最終書き込み者優先の既定のパスは、タイムスタンプフィールドまたは _ts プロパティです。 NoSQL 用 API の場合は、これを数値型のユーザー定義パスに設定することもできます。競合した場合には、最大値が優先されます。このパスが設定されていない場合または無効な場合は、_ts が既定値になります。このポリシーを使用して解決された競合は、競合フィードに表示されません。このポリシーはすべての API で使用できます。

DocumentCollection lwwCollection = await createClient.CreateDocumentCollectionIfNotExistsAsync(
  UriFactory.CreateDatabaseUri(this.databaseName), new DocumentCollection
  {
      Id = this.lwwCollectionName,
      ConflictResolutionPolicy = new ConflictResolutionPolicy
      {
          Mode = ConflictResolutionMode.LastWriterWins,
          ConflictResolutionPath = "/myCustomId",
      },
  });

Container container = await createClient.GetDatabase(this.databaseName)
    .CreateContainerIfNotExistsAsync(new ContainerProperties(this.lwwCollectionName, "/partitionKey")
    {
        ConflictResolutionPolicy = new ConflictResolutionPolicy()
        {
            Mode = ConflictResolutionMode.LastWriterWins,
            ResolutionPath = "/myCustomId",
        }
    });

Java V4 SDK

非同期
同期

Java SDK V4 (Maven com.azure::azure-cosmos) 非同期 API


ConflictResolutionPolicy policy = ConflictResolutionPolicy.createLastWriterWinsPolicy("/myCustomId");

CosmosContainerProperties containerProperties = new CosmosContainerProperties(container_id, partition_key);
containerProperties.setConflictResolutionPolicy(policy);
/* ...other container config... */
database.createContainerIfNotExists(containerProperties).block();

Java SDK V4 (Maven com.azure::azure-cosmos) 同期 API


ConflictResolutionPolicy policy = ConflictResolutionPolicy.createLastWriterWinsPolicy("/myCustomId");

CosmosContainerProperties containerProperties = new CosmosContainerProperties(container_id, partition_key);
containerProperties.setConflictResolutionPolicy(policy);
/* ...other container config... */
database.createContainerIfNotExists(containerProperties);

Java V2 SDK

Async Java V2 SDK
Sync Java V2 SDK

Async Java V2 SDK (Maven com.microsoft.azure::azure-cosmosdb)

DocumentCollection collection = new DocumentCollection();
collection.setId(id);
ConflictResolutionPolicy policy = ConflictResolutionPolicy.createLastWriterWinsPolicy("/myCustomId");
collection.setConflictResolutionPolicy(policy);
DocumentCollection createdCollection = client.createCollection(databaseUri, collection, null).toBlocking().value();

(Maven com.microsoft.azure::azure-documentdb)

DocumentCollection lwwCollection = new DocumentCollection();
lwwCollection.setId(this.lwwCollectionName);
ConflictResolutionPolicy lwwPolicy = ConflictResolutionPolicy.createLastWriterWinsPolicy("/myCustomId");
lwwCollection.setConflictResolutionPolicy(lwwPolicy);
DocumentCollection createdCollection = this.tryCreateDocumentCollection(createClient, database, lwwCollection);

Node.js/JavaScript/TypeScript SDK

const database = client.database(this.databaseName);
const { container: lwwContainer } = await database.containers.createIfNotExists(
  {
    id: this.lwwContainerName,
    conflictResolutionPolicy: {
      mode: "LastWriterWins",
      conflictResolutionPath: "/myCustomId"
    }
  }
);

Python SDK

database = client.get_database_client(database=database_id)
lww_conflict_resolution_policy = {'mode': 'LastWriterWins', 'conflictResolutionPath': '/regionId'}
lww_container = database.create_container(id=lww_container_id, partition_key=PartitionKey(path="/id"), 
    conflict_resolution_policy=lww_conflict_resolution_policy)

ストアドプロシージャを使用したカスタム競合解決ポリシーを作成する

以下のサンプルでは、カスタム競合解決ポリシーを使用してコンテナーを設定する方法について説明します。このポリシーでは、ストアドプロシージャのロジックを使用して競合を解決します。競合を解決するためにストアドプロシージャが指定されている場合、指定されたストアドプロシージャにエラーがない限り、競合フィードに競合は表示されません。

このポリシーがコンテナーを使用して作成された後、ストアドプロシージャを作成する必要があります。以下の .NET SDK サンプルは、このワークフローの例を示します。このポリシーは、NoSQL 用 API でのみサポートされています。

カスタム競合解決ストアドプロシージャの例

カスタム競合解決ストアドプロシージャは、以下に示す関数シグネチャを使用して実装する必要があります。この関数名は、ストアドプロシージャをコンテナーに登録したときに使用された名前と一致する必要はありませんが、名前付けを簡略化します。このストアドプロシージャに実装する必要があるパラメーターの説明を次に示します。

incomingItem: 競合が発生している、コミット時に挿入または更新される項目。削除操作の場合は null 値になります。
existingItem: 現在コミットされている項目。この値は、更新では null 以外の値、挿入または削除では null 値になります。
isTombstone: incomingItem が以前に削除した項目と競合しているかどうかを示すブール値。 true の場合は existingItem も null 値になります。
conflictingItems: ID またはその他の一意なインデックスプロパティが incomingItem と競合している、コンテナー内のすべての項目のコミットされたバージョンによる配列。

重要

ストアドプロシージャと同様、カスタム競合解決プロシージャは、同じパーティションキーを使用してすべてのデータにアクセスすることも、挿入、更新、または削除操作を実行して競合を解決することもできます。

このストアドプロシージャの例では、/myCustomId パスから最小値を選択することで競合を解決します。

function resolver(incomingItem, existingItem, isTombstone, conflictingItems) {
  var collection = getContext().getCollection();

  if (!incomingItem) {
      if (existingItem) {

          collection.deleteDocument(existingItem._self, {}, function (err, responseOptions) {
              if (err) throw err;
          });
      }
  } else if (isTombstone) {
      // delete always wins.
  } else {
      if (existingItem) {
          if (incomingItem.myCustomId > existingItem.myCustomId) {
              return; // existing item wins
          }
      }

      var i;
      for (i = 0; i < conflictingItems.length; i++) {
          if (incomingItem.myCustomId > conflictingItems[i].myCustomId) {
              return; // existing conflict item wins
          }
      }

      // incoming item wins - clear conflicts and replace existing with incoming.
      tryDelete(conflictingItems, incomingItem, existingItem);
  }

  function tryDelete(documents, incoming, existing) {
      if (documents.length > 0) {
          collection.deleteDocument(documents[0]._self, {}, function (err, responseOptions) {
              if (err) throw err;

              documents.shift();
              tryDelete(documents, incoming, existing);
          });
      } else if (existing) {
          collection.replaceDocument(existing._self, incoming,
              function (err, documentCreated) {
                  if (err) throw err;
              });
      } else {
          collection.createDocument(collection.getSelfLink(), incoming,
              function (err, documentCreated) {
                  if (err) throw err;
              });
      }
  }
}

.NET SDK

.NET SDK V2
.NET SDK V3

DocumentCollection udpCollection = await createClient.CreateDocumentCollectionIfNotExistsAsync(
  UriFactory.CreateDatabaseUri(this.databaseName), new DocumentCollection
  {
      Id = this.udpCollectionName,
      ConflictResolutionPolicy = new ConflictResolutionPolicy
      {
          Mode = ConflictResolutionMode.Custom,
          ConflictResolutionProcedure = string.Format("dbs/{0}/colls/{1}/sprocs/{2}", this.databaseName, this.udpCollectionName, "resolver"),
      },
  });

//Create the stored procedure
await clients[0].CreateStoredProcedureAsync(
UriFactory.CreateStoredProcedureUri(this.databaseName, this.udpCollectionName, "resolver"), new StoredProcedure
{
    Id = "resolver",
    Body = File.ReadAllText(@"resolver.js")
});

Container container = await createClient.GetDatabase(this.databaseName)
    .CreateContainerIfNotExistsAsync(new ContainerProperties(this.udpCollectionName, "/partitionKey")
    {
        ConflictResolutionPolicy = new ConflictResolutionPolicy()
        {
            Mode = ConflictResolutionMode.Custom,
            ResolutionProcedure = string.Format("dbs/{0}/colls/{1}/sprocs/{2}", this.databaseName, this.udpCollectionName, "resolver")
        }
    });

await container.Scripts.CreateStoredProcedureAsync(
    new StoredProcedureProperties("resolver", File.ReadAllText(@"resolver.js"))
);

Java V4 SDK

非同期
同期

Java SDK V4 (Maven com.azure::azure-cosmos) 非同期 API


ConflictResolutionPolicy policy = ConflictResolutionPolicy.createCustomPolicy("resolver");

CosmosContainerProperties containerProperties = new CosmosContainerProperties(container_id, partition_key);
containerProperties.setConflictResolutionPolicy(policy);
/* ...other container config... */
database.createContainerIfNotExists(containerProperties).block();

Java SDK V4 (Maven com.azure::azure-cosmos) 同期 API


ConflictResolutionPolicy policy = ConflictResolutionPolicy.createCustomPolicy("resolver");

CosmosContainerProperties containerProperties = new CosmosContainerProperties(container_id, partition_key);
containerProperties.setConflictResolutionPolicy(policy);
/* ...other container config... */
database.createContainerIfNotExists(containerProperties);

Java V2 SDK

Async Java V2 SDK
Sync Java V2 SDK

Async Java V2 SDK (Maven com.microsoft.azure::azure-cosmosdb)

DocumentCollection collection = new DocumentCollection();
collection.setId(id);
ConflictResolutionPolicy policy = ConflictResolutionPolicy.createCustomPolicy("resolver");
collection.setConflictResolutionPolicy(policy);
DocumentCollection createdCollection = client.createCollection(databaseUri, collection, null).toBlocking().value();

(Maven com.microsoft.azure::azure-documentdb)

DocumentCollection udpCollection = new DocumentCollection();
udpCollection.setId(this.udpCollectionName);
ConflictResolutionPolicy udpPolicy = ConflictResolutionPolicy.createCustomPolicy(
        String.format("dbs/%s/colls/%s/sprocs/%s", this.databaseName, this.udpCollectionName, "resolver"));
udpCollection.setConflictResolutionPolicy(udpPolicy);
DocumentCollection createdCollection = this.tryCreateDocumentCollection(createClient, database, udpCollection);

コンテナーが作成されたら、resolver ストアドプロシージャを作成する必要があります。

Node.js/JavaScript/TypeScript SDK

const database = client.database(this.databaseName);
const { container: udpContainer } = await database.containers.createIfNotExists(
  {
    id: this.udpContainerName,
    conflictResolutionPolicy: {
      mode: "Custom",
      conflictResolutionProcedure: `dbs/${this.databaseName}/colls/${
        this.udpContainerName
      }/sprocs/resolver`
    }
  }
);

コンテナーが作成されたら、resolver ストアドプロシージャを作成する必要があります。

Python SDK

database = client.get_database_client(database=database_id)
udp_custom_resolution_policy = {'mode': 'Custom' }
udp_container = database.create_container(id=udp_container_id, partition_key=PartitionKey(path="/id"),
    conflict_resolution_policy=udp_custom_resolution_policy)

コンテナーが作成されたら、resolver ストアドプロシージャを作成する必要があります。

カスタム競合解決ポリシーを作成する

以下のサンプルでは、カスタム競合解決ポリシーを使用してコンテナーを設定する方法について説明します。この実装では、競合フィードに各競合が表示されます。競合フィードから競合を個別に処理するのはお客様です。

.NET SDK

.NET SDK V2
.NET SDK V3

DocumentCollection manualCollection = await createClient.CreateDocumentCollectionIfNotExistsAsync(
  UriFactory.CreateDatabaseUri(this.databaseName), new DocumentCollection
  {
      Id = this.manualCollectionName,
      ConflictResolutionPolicy = new ConflictResolutionPolicy
      {
          Mode = ConflictResolutionMode.Custom,
      },
  });

Container container = await createClient.GetDatabase(this.databaseName)
    .CreateContainerIfNotExistsAsync(new ContainerProperties(this.manualCollectionName, "/partitionKey")
    {
        ConflictResolutionPolicy = new ConflictResolutionPolicy()
        {
            Mode = ConflictResolutionMode.Custom
        }
    });

Java V4 SDK

非同期
同期

Java SDK V4 (Maven com.azure::azure-cosmos) 非同期 API


ConflictResolutionPolicy policy = ConflictResolutionPolicy.createCustomPolicy();

CosmosContainerProperties containerProperties = new CosmosContainerProperties(container_id, partition_key);
containerProperties.setConflictResolutionPolicy(policy);
/* ...other container config... */
database.createContainerIfNotExists(containerProperties).block();

Java SDK V4 (Maven com.azure::azure-cosmos) 同期 API


ConflictResolutionPolicy policy = ConflictResolutionPolicy.createCustomPolicy();

CosmosContainerProperties containerProperties = new CosmosContainerProperties(container_id, partition_key);
containerProperties.setConflictResolutionPolicy(policy);
/* ...other container config... */
database.createContainerIfNotExists(containerProperties);

Java V2 SDK

Async Java V2 SDK
Sync Java V2 SDK

Async Java V2 SDK (Maven com.microsoft.azure::azure-cosmosdb)

DocumentCollection collection = new DocumentCollection();
collection.setId(id);
ConflictResolutionPolicy policy = ConflictResolutionPolicy.createCustomPolicy();
collection.setConflictResolutionPolicy(policy);
DocumentCollection createdCollection = client.createCollection(databaseUri, collection, null).toBlocking().value();

(Maven com.microsoft.azure::azure-documentdb)

DocumentCollection manualCollection = new DocumentCollection();
manualCollection.setId(this.manualCollectionName);
ConflictResolutionPolicy customPolicy = ConflictResolutionPolicy.createCustomPolicy(null);
manualCollection.setConflictResolutionPolicy(customPolicy);
DocumentCollection createdCollection = client.createCollection(database.getSelfLink(), collection, null).getResource();

Node.js/JavaScript/TypeScript SDK

const database = client.database(this.databaseName);
const {
  container: manualContainer
} = await database.containers.createIfNotExists({
  id: this.manualContainerName,
  conflictResolutionPolicy: {
    mode: "Custom"
  }
});

Python SDK

database = client.get_database_client(database=database_id)
manual_resolution_policy = {'mode': 'Custom'}
manual_container = database.create_container(id=manual_container_id, partition_key=PartitionKey(path="/id"), 
    conflict_resolution_policy=manual_resolution_policy)

競合フィードから読み取りを行う

以下のサンプルでは、コンテナーの競合フィードから読み取りを行う方法について説明します。競合は、いくつかの理由で競合フィードのみに表示される場合があります。

競合が自動的に解決されなかった
競合により、指定されたストアドプロシージャでエラーが発生した
競合解決ポリシーがカスタムに設定され、競合を処理するストアドプロシージャが指定されていない

.NET SDK

.NET SDK V2
.NET SDK V3

FeedResponse<Conflict> conflicts = await delClient.ReadConflictFeedAsync(this.collectionUri);

FeedIterator<ConflictProperties> conflictFeed = container.Conflicts.GetConflictQueryIterator();
while (conflictFeed.HasMoreResults)
{
    FeedResponse<ConflictProperties> conflicts = await conflictFeed.ReadNextAsync();
    foreach (ConflictProperties conflict in conflicts)
    {
        // Read the conflicted content
        MyClass intendedChanges = container.Conflicts.ReadConflictContent<MyClass>(conflict);
        MyClass currentState = await container.Conflicts.ReadCurrentAsync<MyClass>(conflict, new PartitionKey(intendedChanges.MyPartitionKey));

        // Do manual merge among documents
        await container.ReplaceItemAsync<MyClass>(intendedChanges, intendedChanges.Id, new PartitionKey(intendedChanges.MyPartitionKey));

        // Delete the conflict
        await container.Conflicts.DeleteAsync(conflict, new PartitionKey(intendedChanges.MyPartitionKey));
    }
}

Java SDK

Java V4 SDK (Maven com.azure::azure-cosmos)

int requestPageSize = 3;
CosmosQueryRequestOptions options = new CosmosQueryRequestOptions();

CosmosPagedFlux<CosmosConflictProperties> conflictReadFeedFlux = container.readAllConflicts(options);

conflictReadFeedFlux.byPage(requestPageSize).toIterable().forEach(page -> {

    int expectedNumberOfConflicts = 0;
    int numberOfResults = 0;
    Iterator<CosmosConflictProperties> pageIt = page.getElements().iterator();

    while (pageIt.hasNext()) {
        CosmosConflictProperties conflictProperties = pageIt.next();

        // Read the conflict and committed item
        CosmosAsyncConflict conflict = container.getConflict(conflictProperties.getId());
        CosmosConflictResponse response = conflict.read(new CosmosConflictRequestOptions()).block();

        // response.
    }
});

Async Java V2 SDK (Maven com.microsoft.azure::azure-cosmosdb)

FeedResponse<Conflict> response = client.readConflicts(this.manualCollectionUri, null)
                    .first().toBlocking().single();
for (Conflict conflict : response.getResults()) {
    /* Do something with conflict */
}

(Maven com.microsoft.azure::azure-documentdb)

Iterator<Conflict> conflictsIterator = client.readConflicts(this.collectionLink, null).getQueryIterator();
while (conflictsIterator.hasNext()) {
    Conflict conflict = conflictsIterator.next();
    /* Do something with conflict */
}

Node.js/JavaScript/TypeScript SDK

const container = client
  .database(this.databaseName)
  .container(this.lwwContainerName);

const { result: conflicts } = await container.conflicts.readAll().toArray();

Python

conflicts_iterator = iter(container.list_conflicts())
conflict = next(conflicts_iterator, None)
while conflict:
    # Do something with conflict
    conflict = next(conflicts_iterator, None)

次のステップ

Azure Cosmos DB の次の概念について学習しましょう。

次の方法で共有

Azure Cosmos DB での競合解決ポリシーの管理

最終書き込み者優先競合解決ポリシーを作成する

.NET SDK

Java V4 SDK

Java V2 SDK

Node.js/JavaScript/TypeScript SDK

Python SDK

ストアドプロシージャを使用したカスタム競合解決ポリシーを作成する

カスタム競合解決ストアドプロシージャの例

.NET SDK

Java V4 SDK

Java V2 SDK

Node.js/JavaScript/TypeScript SDK

Python SDK

カスタム競合解決ポリシーを作成する

.NET SDK

Java V4 SDK

Java V2 SDK

Node.js/JavaScript/TypeScript SDK

Python SDK

競合フィードから読み取りを行う

.NET SDK

Java SDK

Node.js/JavaScript/TypeScript SDK

Python

次のステップ

フィードバック

その他のリソース

次の方法で共有

Azure Cosmos DB での競合解決ポリシーの管理

最終書き込み者優先競合解決ポリシーを作成する

.NET SDK

Java V4 SDK

Java V2 SDK

Node.js/JavaScript/TypeScript SDK

Python SDK

ストアド プロシージャを使用したカスタム競合解決ポリシーを作成する

カスタム競合解決ストアド プロシージャの例

.NET SDK

Java V4 SDK

Java V2 SDK

Node.js/JavaScript/TypeScript SDK

Python SDK

カスタム競合解決ポリシーを作成する

.NET SDK

Java V4 SDK

Java V2 SDK

Node.js/JavaScript/TypeScript SDK

Python SDK

競合フィードから読み取りを行う

.NET SDK

Java SDK

Node.js/JavaScript/TypeScript SDK

Python

次のステップ

フィードバック

その他のリソース

ストアドプロシージャを使用したカスタム競合解決ポリシーを作成する

カスタム競合解決ストアドプロシージャの例