Transactions Cloud Datastore

Une transaction est un ensemble d'opérations Datastore sur une ou plusieurs entités dans un maximum de 25 groupes d'entités. Chaque transaction est atomique, ce qui signifie que les transactions ne sont jamais partiellement appliquées. Les opérations de transaction s'appliquent toutes, ou aucune ne s'applique.

Utiliser des transactions

Les transactions ont une durée maximale de 270 secondes avec un délai d'expiration inactif de 10 secondes au bout de 30 secondes.

Une opération peut échouer lorsque :

trop de modifications simultanées s'effectuent sur le même groupe d'entités ;
la transaction dépasse une limite de ressources ;
Datastore rencontre une erreur interne.

Dans tous les cas précédents, l'API Datastore renvoie une erreur.

Les transactions représentent une fonctionnalité facultative de Datastore. Vous n'êtes pas obligé de les utiliser pour effectuer des opérations Datastore.

Une application peut exécuter un ensemble d'instructions et d'opérations Datastore en une seule transaction. Ainsi, si une instruction ou une opération déclenche une exception, aucune des opérations Datastore de l’ensemble n’est appliquée. L'application définit les actions à effectuer dans la transaction.

L'extrait de code suivant montre comment effectuer une transaction à l'aide de l'API Datastore. Il transfère de l'argent d'un compte à un autre.

C#

Pour savoir comment installer et utiliser la bibliothèque cliente pour Cloud Datastore, consultez la page Bibliothèques clientes Cloud Datastore. Pour plus d'informations, consultez la documentation de référence de l'API Cloud Datastore C#.

Pour vous authentifier auprès de Cloud Datastore, configurez les Identifiants par défaut de l'application. Pour en savoir plus, consultez Configurer l'authentification pour un environnement de développement local.

private void TransferFunds(Key fromKey, Key toKey, long amount)
{
    using (var transaction = _db.BeginTransaction())
    {
        var entities = transaction.Lookup(fromKey, toKey);
        entities[0]["balance"].IntegerValue -= amount;
        entities[1]["balance"].IntegerValue += amount;
        transaction.Update(entities);
        transaction.Commit();
    }
}

Go

type BankAccount struct {
	Balance int
}

const amount = 50
keys := []*datastore.Key{to, from}
tx, err := client.NewTransaction(ctx)
if err != nil {
	log.Fatalf("client.NewTransaction: %v", err)
}
accs := make([]BankAccount, 2)
if err := tx.GetMulti(keys, accs); err != nil {
	tx.Rollback()
	log.Fatalf("tx.GetMulti: %v", err)
}
accs[0].Balance += amount
accs[1].Balance -= amount
if _, err := tx.PutMulti(keys, accs); err != nil {
	tx.Rollback()
	log.Fatalf("tx.PutMulti: %v", err)
}
if _, err = tx.Commit(); err != nil {
	log.Fatalf("tx.Commit: %v", err)
}

Java

void transferFunds(Key fromKey, Key toKey, long amount) {
  Transaction txn = datastore.newTransaction();
  try {
    List<Entity> entities = txn.fetch(fromKey, toKey);
    Entity from = entities.get(0);
    Entity updatedFrom =
        Entity.newBuilder(from).set("balance", from.getLong("balance") - amount).build();
    Entity to = entities.get(1);
    Entity updatedTo =
        Entity.newBuilder(to).set("balance", to.getLong("balance") + amount).build();
    txn.put(updatedFrom, updatedTo);
    txn.commit();
  } finally {
    if (txn.isActive()) {
      txn.rollback();
    }
  }
}

Node.js

async function transferFunds(fromKey, toKey, amount) {
  const transaction = datastore.transaction();
  await transaction.run();
  const results = await Promise.all([
    transaction.get(fromKey),
    transaction.get(toKey),
  ]);
  const accounts = results.map(result => result[0]);

  accounts[0].balance -= amount;
  accounts[1].balance += amount;

  transaction.save([
    {
      key: fromKey,
      data: accounts[0],
    },
    {
      key: toKey,
      data: accounts[1],
    },
  ]);

  return await transaction.commit();
}

PHP

/**
 * Update two entities in a transaction.
 *
 * @param DatastoreClient $datastore
 * @param Key $fromKey
 * @param Key $toKey
 * @param $amount
 */
function transfer_funds(
    DatastoreClient $datastore,
    Key $fromKey,
    Key $toKey,
    $amount
) {
    $transaction = $datastore->transaction();
    // The option 'sort' is important here, otherwise the order of the result
    // might be different from the order of the keys.
    $result = $transaction->lookupBatch([$fromKey, $toKey], ['sort' => true]);
    if (count($result['found']) != 2) {
        $transaction->rollback();
    }
    $fromAccount = $result['found'][0];
    $toAccount = $result['found'][1];
    $fromAccount['balance'] -= $amount;
    $toAccount['balance'] += $amount;
    $transaction->updateBatch([$fromAccount, $toAccount]);
    $transaction->commit();
}

Python

from google.cloud import datastore

# For help authenticating your client, visit
# https://cloud.google.com/docs/authentication/getting-started
client = datastore.Client()

def transfer_funds(client, from_key, to_key, amount):
    with client.transaction():
        from_account = client.get(from_key)
        to_account = client.get(to_key)

        from_account["balance"] -= amount
        to_account["balance"] += amount

        client.put_multi([from_account, to_account])

Ruby

def transfer_funds from_key, to_key, amount
  datastore.transaction do |tx|
    from = tx.find from_key
    from["balance"] -= amount
    to = tx.find to_key
    to["balance"] += amount
    tx.save from, to
  end
end

Veuillez noter que, pour la clarté de nos exemples, nous omettons parfois la commande rollback en cas d'échec de la transaction. Dans le code de production, il est important de s'assurer que chaque transaction est validée ou annulée de façon explicite.

Que peut-on faire dans une transaction ?

Toutes les opérations Datastore d'une transaction peuvent s'effectuer sur un maximum de 25 groupes d'entités. Cela inclut l'interrogation d'entités par ancêtre, la récupération d'entités par clé, la mise à jour et la suppression d'entités.

Lorsqu'au moins deux transactions tentent simultanément de modifier des entités dans un ou plusieurs groupes d'entités communs, seule la première transaction à valider ses modifications peut aboutir. Toutes les autres transactions échoueront au moment du commit. En raison de cette conception, l'utilisation de groupes d'entités limite le nombre d'écritures simultanées qu'il est possible d'effectuer sur n'importe quelle entité des groupes. Lorsqu'une transaction démarre, Datastore utilise un contrôle de simultanéité optimiste en vérifiant l'heure de la dernière mise à jour des groupes d'entités utilisés dans la transaction. Lors du commit d'une transaction pour les groupes d'entités, Datastore vérifie à nouveau l'heure de la dernière mise à jour pour les groupes d'entités utilisés dans la transaction. Si elle a changé depuis la première vérification, une erreur est renvoyée. Pour en savoir plus sur les groupes d'entités, reportez-vous à la section Chemins d'ancêtres.

Isolation et cohérence

En dehors des transactions, le niveau d'isolation de Datastore est le plus proche du niveau Read committed (lecture de données validées). À l'intérieur des transactions, c'est l'isolation sérialisable qui est appliquée. Cela signifie qu'il est impossible pour une autre transaction de modifier simultanément les données lues ou modifiées par cette transaction. Lisez le wiki sur l'isolation sérialisable (en anglais) et l'article sur l'isolation de transaction pour en savoir plus sur les niveaux d'isolation.

Dans une transaction, toutes les opérations de lecture reflètent l'état actuel et cohérent de Datastore au moment du démarrage de la transaction. Les requêtes et les recherches au sein d'une transaction sont certaines d'afficher un seul instantané cohérent de Datastore au début de la transaction. Les entités et les lignes d'index dans les groupes d'entités de la transaction sont entièrement mises à jour afin que les requêtes renvoient l'ensemble complet et correct des entités de résultat, sans les faux positifs ou les faux négatifs décrits dans la section Isolation de transaction qui peuvent survenir dans des requêtes en dehors des transactions.

Cette capture d'instantané cohérente s'étend également aux opérations de lecture ayant lieu après les opérations d'écriture au sein des transactions. Contrairement à la plupart des bases de données, les requêtes et les opérations get au sein d'une transaction Datastore n'affichent pas les résultats des écritures précédentes dans cette transaction. Plus précisément, si une entité est modifiée ou supprimée au sein d'une transaction, une requête ou une recherche renvoie la version d'origine de l'entité au début de la transaction ou ne renvoie rien si l'entité n'existait pas à ce moment-là.

Possibilités d'utilisation des transactions

Une utilisation des transactions consiste à mettre à jour une entité avec une nouvelle valeur de propriété par rapport à sa valeur actuelle. L'exemple transferFunds ci-dessus effectue cette action pour deux entités, en retirant de l'argent d'un compte et en le transférant sur un autre. L'API Cloud Datastore n'effectue pas de nouvelles tentatives de transactions. Toutefois, vous pouvez ajouter votre propre logique dans ce domaine, par exemple pour gérer les conflits lorsqu'une autre requête met à jour la même entité au même moment.

C#

/// <summary>
/// Retry the action when a Grpc.Core.RpcException is thrown.
/// </summary>
private T RetryRpc<T>(Func<T> action)
{
    List<Grpc.Core.RpcException> exceptions = null;
    var delayMs = _retryDelayMs;
    for (int tryCount = 0; tryCount < _retryCount; ++tryCount)
    {
        try
        {
            return action();
        }
        catch (Grpc.Core.RpcException e)
        {
            if (exceptions == null)
                exceptions = new List<Grpc.Core.RpcException>();
            exceptions.Add(e);
        }
        System.Threading.Thread.Sleep(delayMs);
        delayMs *= 2;  // Exponential back-off.
    }
    throw new AggregateException(exceptions);
}

private void RetryRpc(Action action)
{
    RetryRpc(() => { action(); return 0; });
}

[Fact]
public void TestTransactionalRetry()
{
    int tryCount = 0;
    var keys = UpsertBalances();
    RetryRpc(() =>
    {
        using (var transaction = _db.BeginTransaction())
        {
            TransferFunds(keys[0], keys[1], 10, transaction);
            // Insert a conflicting transaction on the first try.
            if (tryCount++ == 0)
                TransferFunds(keys[1], keys[0], 5);
            transaction.Commit();
        }
    });
    Assert.Equal(2, tryCount);
}

Go

type BankAccount struct {
	Balance int
}

const amount = 50
_, err := client.RunInTransaction(ctx, func(tx *datastore.Transaction) error {
	keys := []*datastore.Key{to, from}
	accs := make([]BankAccount, 2)
	if err := tx.GetMulti(keys, accs); err != nil {
		return err
	}
	accs[0].Balance += amount
	accs[1].Balance -= amount
	_, err := tx.PutMulti(keys, accs)
	return err
})

Java

int retries = 5;
while (true) {
  try {
    transferFunds(fromKey, toKey, 10);
    break;
  } catch (DatastoreException e) {
    if (retries == 0) {
      throw e;
    }
    --retries;
  }
}
// Retry handling can also be configured and automatically applied using google-cloud-java.

Node.js

async function transferFundsWithRetry() {
  const maxTries = 5;

  async function tryRequest(currentAttempt, delay) {
    try {
      await transferFunds(fromKey, toKey, 10);
    } catch (err) {
      if (currentAttempt <= maxTries) {
        // Use exponential backoff
        setTimeout(async () => {
          await tryRequest(currentAttempt + 1, delay * 2);
        }, delay);
      }
      throw err;
    }
  }

  await tryRequest(1, 100);
}

PHP

$retries = 5;
for ($i = 0; $i < $retries; $i++) {
    try {
        transfer_funds($datastore, $fromKey, $toKey, 10);
    } catch (\Google\Cloud\Core\Exception\ConflictException $e) {
        // if $i >= $retries, the failure is final
        continue;
    }
    // Succeeded!
    break;
}

Python

from google.cloud import datastore

# For help authenticating your client, visit
# https://cloud.google.com/docs/authentication/getting-started
client = datastore.Client()

for _ in range(5):
    try:
        transfer_funds(client, account1.key, account2.key, 50)
        break
    except google.cloud.exceptions.Conflict:
        continue
else:
    print("Transaction failed.")

Ruby

(1..5).each do |i|
  begin
    return transfer_funds from_key, to_key, amount
  rescue Google::Cloud::Error => e
    raise e if i == 5
  end
end

Ici, une transaction est nécessaire, car la valeur balance dans une entité peut être mise à jour par un autre utilisateur après que ce code a récupéré l'objet. Toutefois, cette opération doit s'effectuer avant qu'il n'enregistre l'objet modifié. En l'absence de transaction, la requête de l'utilisateur utilise la valeur balance avant la mise à jour effectuée par l'autre utilisateur, puis la sauvegarde écrase la nouvelle valeur. Avec une transaction, l'application est prévenue de la mise à jour effectuée par l'autre utilisateur.

Une autre utilisation courante des transactions consiste à récupérer une entité avec une clé nommée ou de la créer si elle n'existe pas (cet exemple se base sur l'exemple TaskList illustré dans la section Créer une entité) :

C#

Entity task;
using (var transaction = _db.BeginTransaction())
{
    task = transaction.Lookup(_sampleTask.Key);
    if (task == null)
    {
        transaction.Insert(_sampleTask);
        transaction.Commit();
    }
}

Go

_, err := client.RunInTransaction(ctx, func(tx *datastore.Transaction) error {
	var task Task
	if err := tx.Get(key, &task); err != datastore.ErrNoSuchEntity {
		return err
	}
	_, err := tx.Put(key, &Task{
		Category:    "Personal",
		Done:        false,
		Priority:    4,
		Description: "Learn Cloud Datastore",
	})
	return err
})

Java

Entity task;
Transaction txn = datastore.newTransaction();
try {
  task = txn.get(taskKey);
  if (task == null) {
    task = Entity.newBuilder(taskKey).build();
    txn.put(task);
    txn.commit();
  }
} finally {
  if (txn.isActive()) {
    txn.rollback();
  }
}

Node.js

async function getOrCreate(taskKey, taskData) {
  const taskEntity = {
    key: taskKey,
    data: taskData,
  };
  const transaction = datastore.transaction();

  try {
    await transaction.run();
    const [task] = await transaction.get(taskKey);
    if (task) {
      // The task entity already exists.
      await transaction.rollback();
    } else {
      // Create the task entity.
      transaction.save(taskEntity);
      await transaction.commit();
    }
    return taskEntity;
  } catch (err) {
    await transaction.rollback();
  }
}

PHP

$transaction = $datastore->transaction();
$entity = $transaction->lookup($task->key());
if ($entity === null) {
    $entity = $transaction->insert($task);
    $transaction->commit();
}

Python

from google.cloud import datastore

# For help authenticating your client, visit
# https://cloud.google.com/docs/authentication/getting-started
client = datastore.Client()

with client.transaction():
    key = client.key(
        "Task", datetime.datetime.now(tz=datetime.timezone.utc).isoformat()
    )

    task = client.get(key)

    if not task:
        task = datastore.Entity(key)
        task.update({"description": "Example task"})
        client.put(task)

    return task

Ruby

task = nil
datastore.transaction do |tx|
  task = tx.find task_key
  if task.nil?
    task = datastore.entity task_key do |t|
      t["category"] = "Personal"
      t["done"] = false
      t["priority"] = 4
      t["description"] = "Learn Cloud Datastore"
    end
    tx.save task
  end
end

Comme précédemment, une transaction est nécessaire au cas où un autre utilisateur tente de créer ou de mettre à jour une entité avec le même identifiant de chaîne. En l'absence de transaction, si l'entité n'existe pas et que deux utilisateurs essaient de la créer, le second écrase la contribution du premier sans même en être conscient.

Lorsqu'une transaction échoue, vous pouvez demander à l'application de retenter la transaction jusqu'à ce qu'elle réussisse, ou laisser les utilisateurs traiter l'erreur en la propageant au niveau de l'interface utilisateur de l'application. Il n'est pas nécessaire de créer une boucle afin de retenter chaque transaction.

Enfin, vous pouvez utiliser une transaction pour lire un instantané cohérent de Datastore. Cela peut s'avérer utile lorsque plusieurs opérations de lecture sont nécessaires pour afficher une page ou exporter des données qui doivent être cohérentes. Les transactions de ce genre sont souvent appelées transactions en lecture seule, car elles n'effectuent aucune opération d'écriture. Les transactions à groupe unique en lecture seule n'échouent jamais en raison de modifications simultanées. Ainsi, vous n'avez pas besoin d'implémenter des tentatives en cas d'échec. Toutefois, les transactions de groupes d'entités multiples peuvent échouer à cause de modifications simultanées. Par conséquent, celles-ci doivent comprendre plusieurs tentatives.

C#

Entity taskList;
IReadOnlyList<Entity> tasks;
using (var transaction = _db.BeginTransaction(TransactionOptions.CreateReadOnly()))
{
    taskList = transaction.Lookup(taskListKey);
    var query = new Query("Task")
    {
        Filter = Filter.HasAncestor(taskListKey)
    };
    tasks = transaction.RunQuery(query).Entities;
    transaction.Commit();
}

Go

tx, err := client.NewTransaction(ctx, datastore.ReadOnly)
if err != nil {
	log.Fatalf("client.NewTransaction: %v", err)
}
defer tx.Rollback() // Transaction only used for read.

ancestor := datastore.NameKey("TaskList", "default", nil)
query := datastore.NewQuery("Task").Ancestor(ancestor).Transaction(tx)
var tasks []Task
_, err = client.GetAll(ctx, query, &tasks)

Java

Entity taskList;
QueryResults<Entity> tasks;
Transaction txn =
    datastore.newTransaction(
        TransactionOptions.newBuilder().setReadOnly(ReadOnly.newBuilder().build()).build());
try {
  taskList = txn.get(taskListKey);
  Query<Entity> query =
      Query.newEntityQueryBuilder()
          .setKind("Task")
          .setFilter(PropertyFilter.hasAncestor(taskListKey))
          .build();
  tasks = txn.run(query);
  txn.commit();
} finally {
  if (txn.isActive()) {
    txn.rollback();
  }
}

Node.js

async function getTaskListEntities() {
  const transaction = datastore.transaction({readOnly: true});
  try {
    const taskListKey = datastore.key(['TaskList', 'default']);

    await transaction.run();
    const [taskList] = await transaction.get(taskListKey);
    const query = datastore.createQuery('Task').hasAncestor(taskListKey);
    const [taskListEntities] = await transaction.runQuery(query);
    await transaction.commit();
    return [taskList, taskListEntities];
  } catch (err) {
    await transaction.rollback();
  }
}

PHP

$transaction = $datastore->readOnlyTransaction();
$taskListKey = $datastore->key('TaskList', 'default');
$query = $datastore->query()
    ->kind('Task')
    ->hasAncestor($taskListKey);
$result = $transaction->runQuery($query);
$taskListEntities = [];
$num = 0;
/* @var Entity $task */
foreach ($result as $task) {
    $taskListEntities[] = $task;
    $num += 1;
}

Python

from google.cloud import datastore

# For help authenticating your client, visit
# https://cloud.google.com/docs/authentication/getting-started
client = datastore.Client()

with client.transaction(read_only=True):
    task_list_key = client.key("TaskList", "default")

    task_list = client.get(task_list_key)

    query = client.query(kind="Task", ancestor=task_list_key)
    tasks_in_list = list(query.fetch())

    return task_list, tasks_in_list

Ruby

# task_list_name = "default"
task_list_key = datastore.key "TaskList", task_list_name
datastore.read_only_transaction do |tx|
  task_list = tx.find task_list_key
  query = datastore.query("Task").ancestor(task_list)
  tasks_in_list = tx.run query
end

Transactions et groupes d'entités

Un groupe d'entités est un ensemble d'entités connectées par le biais d'un ancêtre à un élément racine commun. L'organisation des données en groupes d'entités peut limiter le nombre de transactions pouvant être effectuées :

Toutes les données auxquelles accède une transaction doivent être contenues dans 25 groupes d'entités au maximum.
Si vous souhaitez employer des requêtes dans une transaction, les données doivent être organisées en groupes d'entités de sorte que vous puissiez spécifier des filtres d'ancêtres qui correspondent aux données appropriées.
Le débit en écriture est limité à environ une transaction par seconde pour un seul groupe d'entités. Cette limite a été définie car Datastore effectue une réplication synchrone sans maître de chaque groupe d'entités sur une vaste zone géographique, afin de fournir une fiabilité et une tolérance aux pannes élevées.

Avertissement : Le dépassement de la limite de débit en écriture d'une transaction par seconde au sein d'un seul groupe d'entités peut entraîner des conflits de lecture et d'écriture sur tous les groupes d'entités impliqués dans vos transactions.

Dans de nombreuses applications, il est acceptable de recourir à la cohérence à terme (à savoir à une requête non ascendante couvrant plusieurs groupes d'entités et pouvant parfois renvoyer des données légèrement obsolètes) lors de l'obtention d'une vue d'ensemble de données non liées, puis de passer à la cohérence forte (requête ascendante ou opération lookup d'une seule entité) lors de la visualisation ou de la modification d'un seul ensemble de données étroitement liées. Dans ces applications, il est généralement recommandé d'utiliser un groupe d'entités distinct pour chaque ensemble de données étroitement liées. Pour plus d'informations, consultez la section Cohérence des données.

Étapes suivantes

Apprenez-en davantage sur les requêtes Datastore.
Apprenez-en davantage sur la cohérence des données dans Datastore.