Hantera principer för konfliktlösning i Azure Cosmos DB

  • Artikel

GÄLLER FÖR: NoSQL

Med skrivningar i flera regioner kan konflikter uppstå när flera klienter skriver till samma objekt. När en konflikt inträffar kan du lösa konflikten med hjälp av olika konfliktlösningsprinciper. I den här artikeln beskrivs hur du hanterar konfliktlösningsprinciper.

Dricks

Konfliktlösningsprincipen kan bara anges när containern skapas och kan inte ändras när containern har skapats.

Skapa en senaste skrivning vinner-konfliktlösningsprincip

De här exemplen visar hur du konfigurerar en container med en senaste skrivning vinner-konfliktlösningsprincip. Standardsökvägen för last-writer-wins är tidsstämpelfältet eller _ts egenskapen . För API för NoSQL kan detta också anges till en användardefinierad sökväg med en numerisk typ. I en konflikt vinner det högsta värdet. Om sökvägen inte har angetts eller om den är ogiltig är den som standard _ts. Konflikter som har lösts med den här principen visas inte i konfliktflödet. Den här principen kan användas av alla API:er.

.NET SDK

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

Java V4 SDK

Java SDK V4 (Maven com.azure::azure-cosmos) Async 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 V2 SDK:er

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();

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)

Skapa en anpassad konfliktlösningsprincip med hjälp av en lagrad procedur

De här exemplen visar hur du konfigurerar en container med en anpassad konfliktlösningsprincip. Den här principen använder logiken i en lagrad procedur för att lösa konflikten. Om en lagrad procedur är avsedd att lösa konflikter visas inte konflikter i konfliktflödet om det inte finns ett fel i den avsedda lagrade proceduren.

När principen har skapats med containern måste du skapa den lagrade proceduren. Exemplet på .NET SDK nedan visar ett exempel på det här arbetsflödet. Den här principen stöds endast i API:et för NoSQL.

Exempel på lagrad procedur för anpassad konfliktlösning

Lagrade procedurer för anpassad konfliktlösning måste implementeras med hjälp av funktionssignaturen som visas nedan. Funktionsnamnet behöver inte matcha namnet som används när du registrerar den lagrade proceduren med containern, men det förenklar namngivning. Här är en beskrivning av de parametrar som måste implementeras för den här lagrade proceduren.

  • incomingItem: Objektet som infogas eller uppdateras i incheckningen som genererar konflikterna. Är null för borttagningsåtgärder.
  • existingItem: Det för närvarande incheckade objektet. Det här värdet är inte null i en uppdatering och null för en infogning eller borttagning.
  • isTombstone: Booleskt värde som anger om incomingItem står i konflikt med ett tidigare borttaget objekt. När det är sant är existingItem också null.
  • conflictingItems: Matrisen för den bekräftade versionen av alla objekt i containern som står i konflikt med incomingItem på ID eller andra unika indexegenskaper.

Viktigt!

Precis som med alla lagrade procedurer kan en anpassad konfliktlösningsprocedur komma åt alla data med samma partitionsnyckel och kan utföra alla infognings-, uppdaterings- eller borttagningsåtgärder för att lösa konflikter.

Den här lagrade exempelproceduren löser konflikter genom att välja det lägsta värdet från /myCustomId sökvägen.

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

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")
});

Java V4 SDK

Java SDK V4 (Maven com.azure::azure-cosmos) Async 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 V2 SDK:er

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();

När containern har skapats måste du skapa den lagrade proceduren 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`
    }
  }
);

När containern har skapats måste du skapa den lagrade proceduren 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)

När containern har skapats måste du skapa den lagrade proceduren resolver.

Skapa en anpassad konfliktlösningsprincip

De här exemplen visar hur du konfigurerar en container med en anpassad konfliktlösningsprincip. Med den här implementeringen visas varje konflikt i konfliktflödet. Det är upp till dig att hantera konflikterna individuellt från konfliktflödet.

.NET SDK

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

Java V4 SDK

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


ConflictResolutionPolicy policy = ConflictResolutionPolicy.createCustomPolicy();

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

Java V2 SDK:er

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();

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)

Läsa från konfliktflödet

De här exemplen visar hur du läser från en containers konfliktflöde. Konflikter kan bara visas i konfliktflödet av några anledningar:

  • Konflikten löstes inte automatiskt
  • Konflikten orsakade ett fel med den avsedda lagrade proceduren
  • Konfliktlösningsprincipen är inställd på anpassad och anger inte någon lagrad procedur för att hantera konflikter

.NET SDK

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

Java-SDK:er

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.
    }
});

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)

Nästa steg

Läs mer om följande Azure Cosmos DB-begrepp: