Управление политиками разрешения конфликтов в Azure Cosmos DB

  • Применяется к: ✅ NoSQL

При записи данных в различных регионах, если несколько клиентов записывают данные в один и тот же элемент, могут возникать конфликты. При возникновении конфликта можно устранить конфликт с помощью разных политик разрешения конфликтов. В этой статье описывается, как управлять политиками разрешения конфликтов.

Tip

Политика разрешения конфликтов может быть указана только во время создания контейнера и не может быть изменена после создания контейнера.

Создайте политику разрешения конфликтов по принципу "последний пишущий побеждает"

В этих примерах показано, как настроить контейнер с политикой разрешения конфликтов последней записи. Путь по умолчанию для last-writer-wins — это поле метки времени либо свойство _ts. Для API NoSQL этот путь также может быть задан пользователем и иметь числовой тип. В конфликте высшая ценность побеждает. Если путь не задан или недействителен, он по умолчанию используется _ts. Конфликты, разрешенные по этой политике, не отображаются в ленте конфликтов. Все API-интерфейсы могут использовать эту политику.

пакет SDK .NET

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

пакет SDK Java версии 4

Java SDK версии 4 (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();

пакеты SDK для Java версии 2

Пакет SDK Async Java версии 2 (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();

пакет SDK Node.js/JavaScript/TypeScript

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

пакет SDK Python

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)

Пакет SDK для GO

db, _ := c.NewDatabase("demo_db")

_, err = db.CreateContainer(context.Background(), azcosmos.ContainerProperties{
	ID: "demo_container",
	PartitionKeyDefinition: azcosmos.PartitionKeyDefinition{
		Paths: []string{"/id"},
		Kind:  azcosmos.PartitionKeyKindHash,
	},
	ConflictResolutionPolicy: &azcosmos.ConflictResolutionPolicy{
		Mode:           azcosmos.ConflictResolutionModeLastWriteWins,
		ResolutionPath: "/myCustomId",
	},
}, nil)

Создание пользовательской политики разрешения конфликтов с помощью хранимой процедуры

В этих примерах показано, как настроить контейнер с помощью настраиваемой политики разрешения конфликтов. Эта политика использует логику в хранимой процедуре для устранения конфликта. Если хранимая процедура назначена для разрешения конфликтов, то конфликты не отображаются в ленте конфликтов, если в указанной хранимой процедуре не возникает ошибка.

После создания политики с контейнером необходимо создать хранимую процедуру. В примере пакета SDK для .NET показан пример этого рабочего процесса. Эта политика поддерживается только в API для NoSQL.

Пример хранимой процедуры разрешения пользовательских конфликтов

Пользовательские хранимые процедуры разрешения конфликтов должны быть реализованы с помощью сигнатуры функции, показанной ниже. Имя функции не обязательно совпадает с именем, используемым при регистрации хранимой процедуры в контейнере, но упрощает именование. Ниже представлено описание параметров, которые должны быть использованы для данной хранимой процедуры.

  • incomingItem: элемент, который вставляется или обновляется в фиксации, которая создает конфликты. Имеет значение NULL для операций удаления.
  • existingItem: текущий зафиксированный элемент. Это значение не равно null при обновлении и равно null при вставке или удалении.
  • isTombstone: логическое значение, указывающее, конфликтует ли входящийItem с ранее удаленным элементом. Если значение true, то существующийItem также имеет значение NULL.
  • conflictingItems: массив зафиксированной версии всех элементов в контейнере, которые конфликтуют со входящими элементами по идентификаторам или другим уникальным свойствам индекса.

Important

Как и любая хранимая процедура, пользовательская процедура разрешения конфликтов может получить доступ к любым данным с одним ключом секции и выполнять любую операцию вставки, обновления или удаления для разрешения конфликтов.

Этот пример процедуры хранения разрешает конфликты, выбирая наименьшее значение в /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;
              });
      }
  }
}

пакет SDK .NET

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

пакет SDK Java версии 4

Java SDK версии 4 (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();

пакеты SDK для Java версии 2

Пакет SDK Async Java версии 2 (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();

После создания контейнера необходимо создать хранимую процедуру resolver.

пакет SDK Node.js/JavaScript/TypeScript

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.

пакет SDK Python

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.

Создание настраиваемой политики разрешения конфликтов

В этих примерах показано, как настроить контейнер с помощью настраиваемой политики разрешения конфликтов. При этой реализации каждый конфликт отображается в канале конфликтов. Ваша задача — обрабатывать конфликты по отдельности из списка конфликтов.

пакет SDK .NET

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

пакет SDK Java версии 4

Java SDK версии 4 (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();

пакеты SDK для Java версии 2

Пакет SDK Async Java версии 2 (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();

пакет SDK Node.js/JavaScript/TypeScript

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

пакет SDK Python

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)

Чтение из канала конфликтов

В этих примерах показано, как считывать из журнала конфликтов контейнера. Конфликты могут отображаться в ленте конфликтов только по нескольким причинам:

  • Конфликт не был разрешен автоматически
  • Конфликт вызвал ошибку с указанной хранимой процедурой
  • Политика разрешения конфликтов настроена на пользовательскую и не назначает хранимую процедуру для обработки конфликтов.

пакет SDK .NET

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

Пакеты SDK для Java

Пакет SDK для Java версии 4 (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.
    }
});

пакет SDK Node.js/JavaScript/TypeScript

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:

  • Last updated on