Cloud Datastore-Transaktionen

Eine Transaktion ist eine Gruppe von Datastore-Vorgängen zu einer oder mehreren Entitäten in bis zu 25 Entitätengruppen. Jede Transaktion ist garantiert atomar, Transaktionen werden also niemals teilweise angewendet. Entweder werden alle Vorgänge in der Transaktion angewendet oder keiner.

Transaktionen verwenden

Transaktionen haben eine maximale Dauer von 270 Sekunden mit einer Inaktivitätsablaufzeit von 10 Sekunden nach 30 Sekunden.

In folgenden Fällen kann ein Vorgang fehlschlagen:

Es wurden zu viele gleichzeitige Änderungen für dieselbe Entitätsgruppe versucht.
Die Transaktion überschreitet einen Ressourcengrenzwert.
Datastore hat einen internen Fehler festgestellt.

In allen diesen Fällen gibt die Datastore API einen Fehler zurück.

Transaktionen sind ein optionales Feature von Datastore. Sie sind nicht verpflichtet, Transaktionen für Datastore-Vorgänge zu verwenden.

In Anwendungen können mehrere Anweisungen und Datastore-Vorgänge in einer einzelnen Transaktion ausgeführt werden. Dies bewirkt, dass keiner der entsprechenden Datastore-Vorgänge angewendet wird, wenn bei einer Anweisung oder einem Vorgang eine Ausnahme ausgegeben wird. Die Anwendung definiert die in der Transaktion auszuführenden Aktionen.

Das folgende Snippet zeigt, wie eine Transaktion mithilfe der Datastore API ausgeführt wird. In diesem Beispiel wird Geld von einem Konto auf ein anderes überwiesen.

C#

Informationen zum Installieren und Verwenden der Clientbibliothek für Cloud Datastore finden Sie unter Cloud Datastore-Clientbibliotheken. Weitere Informationen finden Sie in der Referenzdokumentation zur Cloud Datastore C# API.

Richten Sie Standardanmeldedaten für Anwendungen ein, um sich bei Cloud Datastore zu authentifizieren. Weitere Informationen finden Sie unter Authentifizierung für eine lokale Entwicklungsumgebung einrichten.

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

Um unsere Beispiele möglichst kurz zu halten, lassen wir gelegentlich das rollback weg, wenn die Transaktion fehlschlägt. Bei Produktionscode muss unbedingt dafür gesorgt werden, dass für jede Transaktion explizit entweder ein Commit oder ein Rollback durchgeführt wird.

Möglichkeiten in einer Transaktion

Alle Datastore-Vorgänge in einer Transaktion können für maximal 25 Entitätengruppen durchgeführt werden. Hierzu gehört das Abfragen von Entitäten nach Ancestor, das Abrufen von Entitäten nach Schlüssel, das Aktualisieren von Entitäten und das Löschen von Entitäten.

Wenn zwei oder mehr Transaktionen gleichzeitig versuchen, Entitäten in einer oder mehreren gemeinsamen Entitätsgruppen zu ändern, ist nur die erste Transaktion, die ihre Änderungen mit Commit speichert, erfolgreich. Alle anderen schlagen beim Commit fehl. Aufgrund dieses Designs begrenzt die Verwendung von Entitätengruppen die Anzahl von gleichzeitigen Schreibvorgängen in eine Entität innerhalb der Gruppen. Beim Start einer Transaktion verwendet Datastore die optimistische Nebenläufigkeitserkennung, indem die letzte Aktualisierungszeit für die Entitätengruppen in der Transaktion geprüft wird. Beim Commit einer Transaktion für die Entitätengruppen prüft Datastore noch einmal die letzte Aktualisierungszeit für die Entitätengruppen in der Transaktion. Wenn sich diese seit der anfänglichen Prüfung geändert hat, wird ein Fehler zurückgegeben. Eine Erläuterung der Entitätengruppen finden Sie in Ancestor-Pfade.

Isolation und Konsistenz

Außerhalb von Transaktionen ist die Isolationsebene von Datastore dem Lesevorgang am ähnlichsten, für den ein Commit durchgeführt wurde. Innerhalb von Transaktionen wird die serialisierbare Isolation erzwungen. Dies bedeutet, dass keine andere Transaktion die Daten, die von dieser Transaktion gelesen oder geändert werden, gleichzeitig ändern kann. Weitere Informationen zu Isolationsebenen finden Sie im Artikel zur serialisierbaren Isolation und im Artikel zur Transaktionsisolation.

In einer Transaktion spiegeln alle Lesevorgänge den aktuellen, konsistenten Status von Datastore zu Beginn der Transaktion wider. Abfragen und Suchvorgänge innerhalb einer Transaktion sehen garantiert einen einzelnen konsistenten Snapshot von Datastore seit dem Beginn der Transaktion. Entitäten und Indexzeilen in den Entitätengruppen der Transaktion werden vollständig aktualisiert. So geben Abfragen die vollständige, korrekte Menge von Ergebnisentitäten zurück, ohne fälschlicherweise positive oder fälschlicherweise negative Ergebnisse – wie in Transaktionsisolation beschrieben – zurückzugeben, die in Abfragen außerhalb von Transaktionen auftreten können.

Diese konsistente Snapshot-Ansicht wird innerhalb von Transaktionen auch nach Schreibvorgängen auf Lesevorgänge erweitert. Im Gegensatz zu den meisten Datenbanken sehen Abfragen und Abrufe innerhalb einer Datastore-Transaktion die Ergebnisse vorheriger Schreibvorgänge innerhalb dieser Transaktion nicht. Wenn eine Entität innerhalb einer Transaktion geändert oder gelöscht wird, gibt eine Abfrage oder ein Lookup die ursprüngliche Version der Entität vom Beginn der Transaktion zurück, bzw. nichts, wenn die Entität zu dem Zeitpunkt nicht vorhanden war.

Anwendungsfälle für Transaktionen

Eine Verwendung von Transaktionen besteht in der Aktualisierung einer Entität mit einem neuen Attributwert bezogen auf den aktuellen Wert. Im transferFunds-Beispiel oben wird dies für zwei Entitäten durchgeführt, indem Geldbeträge von einem Konto abgezogen und auf ein anderes überwiesen werden. Die Datastore API wiederholt Transaktionen nicht automatisch. Sie können jedoch Ihre eigene Logik zur Wiederholung dieser Transaktionen hinzufügen, beispielsweise um Konflikte zu lösen, wenn eine andere Anfrage dieselbe Entität gleichzeitig aktualisiert.

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

Dies erfordert eine Transaktion, weil der Wert von balance in einer Entität von einem anderen Nutzer aktualisiert werden kann, nachdem dieser Code das Objekt abgerufen, doch bevor er das geänderte Objekt gespeichert hat. Ohne eine Transaktion verwendet die Anfrage des Nutzers den Wert von balance vor der Aktualisierung durch den anderen Nutzer und beim Speichern wird der neue Wert überschrieben. Mit einer Transaktion wird die Anwendung über die Aktualisierung durch den anderen Nutzer informiert.

Transaktionen werden auch häufig verwendet, um eine Entität mit einem benannten Schlüssel abzurufen oder die Entität zu erstellen, wenn sie noch nicht vorhanden ist (dieses Beispiel baut auf dem TaskList-Beispiel in Entität erstellen auf):

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

Wie im vorherigen Fall ist eine Transaktion erforderlich, um den Fall zu behandeln, dass ein anderer Nutzer gerade versucht, eine Entität mit derselben Zeichenfolge-ID zu erstellen oder zu aktualisieren. Wenn eine Entität nicht existiert und zwei Nutzer gleichzeitig versuchen sie zu erstellen, dann überschreibt ohne Verwendung einer Transaktion der zweite Versuch den ersten, ohne dies zu bemerken.

Wenn eine Transaktion fehlschlägt, können Sie veranlassen, dass die Anwendung die Transaktion so lange wiederholt, bis sie erfolgreich abgeschlossen wurde. Sie können aber auch die Nutzer den Fehler behandeln lassen, indem Sie ihn an die Benutzeroberfläche Ihrer Anwendung weiterleiten. Sie müssen keine Wiederholungsschleife für jede Transaktion erstellen.

Schließlich können Sie eine Transaktion verwenden, um einen konsistenten Snapshot von Datastore zu lesen. Dies kann nützlich sein, wenn mehrere Lesevorgänge erforderlich sind, um eine Seite anzuzeigen oder Daten zu exportieren, die konsistent sein müssen. Diese Arten von Transaktionen werden häufig als schreibgeschützte Transaktionen bezeichnet, weil sie keine Schreibvorgänge ausführen. Schreibgeschützte Einzelgruppentransaktionen schlagen niemals wegen gleichzeitiger Änderungen fehl, sodass Sie nach einem Fehler keine Wiederholungen implementieren müssen. Transaktionen mit mehreren Entitätsgruppen können jedoch wegen gleichzeitiger Änderungen fehlschlagen, sodass für sie Wiederholungen festgelegt werden müssen.

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

Transaktionen und Entitätengruppen

Eine Entitätengruppe besteht aus einer Reihe von Entitäten, die über ihre Herkunft mit einem gemeinsamen Stammelement verbunden sind. Durch die Anordnung von Daten in Entitätengruppen kann begrenzt werden, welche Transaktionen ausgeführt werden können:

Alle Daten, auf die eine Transaktion zugreift, müssen in maximal 25 Entitätengruppen enthalten sein.
Wenn Sie Abfragen innerhalb einer Transaktion verwenden möchten, müssen die Daten so in Entitätengruppen angeordnet werden, dass Sie Ancestor-Filter angeben können, die mit den richtigen Daten übereinstimmen.
Es gibt einen Schreibdurchsatzgrenzwert von ca. einer Transaktion pro Sekunde innerhalb einer einzelnen Entitätengruppe. Diese Begrenzung ist vorhanden, weil Datastore eine synchrone Replikation ohne Master jeder Entitätengruppe über einen breiten geografischen Bereich hinweg durchführt, um hohe Verlässlichkeit und Fehlertoleranz zu ermöglichen.

Warnung: Das Überschreiten des Schreibdurchsatzgrenzwerts von einer Transaktion pro Sekunde in einer einzelnen Entitätengruppe kann zu Lese- und Schreibkonflikten in allen Entitätengruppen führen, die an den Transaktionen beteiligt sind.

In vielen Anwendungen kann es akzeptabel sein, für den Abruf einer breiten Ansicht nicht zusammengehöriger Daten Eventual Consistency zu verwenden, also eine entitätengruppenübergreifende Nicht-Ancestor-Abfrage, die gelegentlich etwas veraltete Daten zurückgeben kann. Anschließend wird dann strikte Konsistenz (eine Ancestor-Abfrage oder ein lookup einer einzelnen Entität) verwendet, um eine einzelne Gruppe von eng zusammengehörigen Daten anzusehen oder zu bearbeiten. In derartigen Anwendungen ist die Gruppierung von eng zusammengehörigen Daten in Entitätengruppen im Allgemeinen ein guter Ansatz. Weitere Informationen finden Sie in Datenkonsistenz.

Nächste Schritte

Weitere Informationen zu Datastore-Abfragen
Weitere Informationen zur Datenkonsistenz in Datastore