并发控制

Spanner 事务提供两种并发控制模式： 悲观和乐观。并发控制模式的选择会影响事务处理同步读取和写入的方式，从而影响性能、延迟时间和事务中止率。请选择最适合您应用的性能和一致性要求的模式。

默认行为取决于事务使用的隔离级别 ：

• 串行化隔离默认使用 悲观并发控制。
• 可重复读隔离， 默认使用乐观并发控制。

悲观并发控制

默认情况下，Spanner 使用悲观并发和 串行化隔离。 您还可以将悲观并发与 可重复读隔离结合使用。

串行化隔离中的悲观并发

此模式假定并发事务可能会争用相同的数据。 它会在事务中读取或写入数据时主动获取数据锁 。它还会验证在事务中较早获取的锁是否在后续语句中保持持有状态。当 Spanner 检测到锁冲突时，它会使用“受伤-等待”算法来解决冲突。

在悲观并发中，事务会在事务的执行阶段和提交阶段获取数据锁。

• 对于读取： 当事务读取数据时，它会在执行阶段获取 共享读取 (ReaderShared) 锁 。这些锁会一直持有到事务提交为止。
• 对于 DML 和写入：
  • 在执行期间，对于由 DML 或写入修改的数据，事务可能会获取行存在性读取锁。
  • 在提交时，事务会尝试获取所写入数据的写入锁或独占锁。写入锁会阻止并发读取，但可能不会阻止并发写入，尤其是在两者都使用写入锁的情况下。这意味着多个事务可以继续提交，并且写入-写入冲突会在提交时使用“受伤-等待”算法解决。所有锁会一直持有到事务提交为止。

可重复读隔离中的悲观并发

在可重复读隔离中使用悲观并发来序列化写入。在此模式下，读取操作使用快照，但 独占锁 适用于从 FOR UPDATE 查询或 lock_scanned_ranges=exclusive 提示读取的数据，以及使用 DML 查询写入的数据。

使用串行化隔离的悲观并发的优势

使用串行化隔离的悲观并发的主要优势在于，在高度争用的工作负载中，它有助于事务取得进展。 Spanner 在冲突期间优先处理较早的事务，而不是较新的事务，从而确保事务最终完成，同时减少重复中止的事务数量。

使用可重复读隔离的悲观并发的优势

使用可重复读隔离时，如果作为 FOR UPDATE 查询的一部分或作为 DML 查询的一部分读取的数据在事务提交之前被并发事务修改，则获取锁的事务可能仍会在提交时中止。不过，在获取锁后，它会阻止进一步的并发更新，直到事务提交，从而序列化写入。

悲观并发的风险

使用串行化隔离的悲观并发存在以下风险：

• 长时间运行的读取可能会阻止对延迟时间敏感的写入。
• 在完成之前涉及用户互动的事务可能会导致锁长时间持有，从而可能会阻止其他操作。

使用串行化隔离的悲观并发的应用场景

悲观并发适用于读写争用和写入-写入争用较高的工作负载。当事务中止和重试成本高昂时，它也适用。除非您的工作负载存在过多的长锁延迟，或者受到锁冲突的严重影响，否则请使用此默认模式。

使用可重复读隔离的悲观并发的应用场景

对于需要 FOR UPDATE 子句或 DML 查询来获取锁的工作负载，请使用可重复读隔离的悲观并发。此方法对于从其他数据库迁移到 Spanner 的工作负载尤其有用，这些数据库会为这些语句获取锁。

乐观并发控制

Spanner 还提供乐观并发控制。当您使用可重复读隔离时，默认模式是乐观并发控制。您还可以将串行化隔离配置为使用乐观并发控制。

乐观并发控制假定冲突很少发生。即使在读写事务中，读取和查询也会在不获取锁的情况下继续进行。 使用 Spanner 的默认串行化隔离时，读取会在提交时进行验证。这可确保没有其他并发提交的事务修改该事务之前读取的数据。如果您使用 可重复读隔离， 带有 FOR UPDATE 或 lock_scanned_ranges=exclusive 提示的读取会在 提交时进行验证。如果 Spanner 检测到冲突，它会中止事务。

乐观并发的工作原理

乐观并发会更改 Spanner 执行读取、查询和提交事务的方式。它会在读取阶段执行无锁执行，并在提交时验证一致性。

对于读取和查询

读取和查询是无锁的。乐观事务中的所有读取和查询都在单个快照时间戳执行。Spanner 会在执行第一个读取或查询时选择此时间戳。这可确保事务中的所有后续读取和查询都会看到在第一个读取或查询之前提交的写入。

对于读取和写入

对于包含读取和写入的乐观事务，Spanner 会在提交时执行验证步骤。只有在未检测到冲突且满足以下条件时，事务才会成功提交：

• 没有并发提交的写入与此事务读取的数据冲突；也就是说，在读取时间戳之后但在该事务提交自己的写入之前，没有提交任何写入。
• 自读取时间戳以来，架构未被修改。

隔离级别决定了要验证的读取集。使用串行化隔离时，所有读取都会经过验证。使用可重复读隔离时，带有 FOR UPDATE 或 lock_scanned_ranges=exclusive 提示的读取会在提交时进行验证。

在高争用情况下，乐观事务可能会重复中止。相比之下，悲观事务通过允许较早的事务提交并重试较新的事务来解决读写冲突。

乐观并发的优势

乐观并发具有以下优势：

• 读取不会获取锁：乐观事务不会为读取获取锁，因此长时间运行的读取不会阻止对延迟时间敏感的写入。
• 缩短只读事务的提交延迟时间：由于乐观事务中的所有读取都基于相同的快照时间戳，因此无需在执行或提交期间验证这些读取的一致性，从而显著缩短了延迟时间。

乐观并发的风险

乐观并发会带来风险，尤其是在与串行化隔离结合使用时，在高读写争用情况下。在为工作负载使用串行化隔离的乐观并发控制之前，请了解这些风险。

• 在高读写争用情况下，乐观事务可能会出现较高的中止率，因为并发写入可能会使乐观事务的读取失效。
• 在持续高争用情况下，事务可能会因事务饥饿而重复中止，并且永远不会提交。

乐观并发的应用场景

乐观并发适用于读写争用较低的事务性工作负载。对于串行化事务，它还有利于可以容忍事务中止的工作负载。

对于以下工作负载，请考虑乐观并发：

• 低优先级、容忍延迟时间且具有长时间运行事务的工作负载： 如果长时间运行的读取或查询可能会延迟对延迟时间敏感的写入，请使用乐观并发。这可以避免因读取锁而导致的延迟。例如，连接速度较慢的移动客户端中的事务，或者为许多行或大范围持有读取锁的低 SLA 事务。
• 对读取延迟时间敏感且读写争用较低的事务性工作负载： 在多区域配置中， 使用乐观并发在区域内提供读取服务，缩短读取延迟时间， 并避免因读取流量激增到热门拆分而导致生产问题。 它还可以提高主要副本过载或不可用期间的读取可用性。
• 大多数事务都是只读的事务性工作负载： 切换到乐观并发可以缩短这些工作负载中常见只读事务的提交延迟时间。确保低读写争用，以避免读写事务出现较高的中止率。

对于读写冲突频繁的对延迟时间敏感的事务性工作负载，请避免使用乐观并发。

配置并发控制

您可以使用 Spanner 客户端库、REST 和 RPC API 来指定读写事务的并发模式。

客户端库

Java

static void readLockModeSetting(DatabaseId db) {
  // The read lock mode specified at the client-level will be applied to all
  // RW transactions.
  DefaultReadWriteTransactionOptions transactionOptions =
      DefaultReadWriteTransactionOptions.newBuilder()
          .setReadLockMode(ReadLockMode.OPTIMISTIC)
          .build();
  SpannerOptions options =
      SpannerOptions.newBuilder()
          .setDefaultTransactionOptions(transactionOptions)
          .build();
  Spanner spanner = options.getService();
  DatabaseClient dbClient = spanner.getDatabaseClient(db);
  dbClient
      // The read lock mode specified at the transaction-level takes precedence
      // over the read lock mode configured at the client-level.
      .readWriteTransaction(Options.readLockMode(ReadLockMode.PESSIMISTIC))
      .run(transaction -> {
        // Read an AlbumTitle.
        String selectSql =
            "SELECT AlbumTitle from Albums WHERE SingerId = 1 and AlbumId = 1";
        String title = null;
        try (ResultSet resultSet = transaction.executeQuery(Statement.of(selectSql))) {
          if (resultSet.next()) {
            title = resultSet.getString("AlbumTitle");
          }
        }
        System.out.printf("Current album title: %s\n", title);

        // Update the title.
        String updateSql =
            "UPDATE Albums "
                + "SET AlbumTitle = 'New Album Title' "
                + "WHERE SingerId = 1 and AlbumId = 1";
        long rowCount = transaction.executeUpdate(Statement.of(updateSql));
        System.out.printf("%d record updated.\n", rowCount);
        return null;
      });
}

Go

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/spanner"
	pb "cloud.google.com/go/spanner/apiv1/spannerpb"
)

// writeWithTransactionUsingReadLockMode sets the ReadLockMode globally
// by using ClientConfig and shows how to override it for a specific
// transaction. ReadLockMode determines the locking strategy used during
// transaction execution.
func writeWithTransactionUsingReadLockMode(w io.Writer, db string) error {
	ctx := context.Background()

	// Client-level configuration: Applies to all read-write transactions
	// for this client. OPTIMISTIC mode avoids locks during reads and
	// verifies changes during the commit phase.
	cfg := spanner.ClientConfig{
		TransactionOptions: spanner.TransactionOptions{
			ReadLockMode: pb.TransactionOptions_ReadWrite_OPTIMISTIC,
		},
	}
	client, err := spanner.NewClientWithConfig(ctx, db, cfg)
	if err != nil {
		return fmt.Errorf("failed to create client: %w", err)
	}
	defer client.Close()

	// Transaction-level options take precedence over client-level
	// configuration. PESSIMISTIC mode is used here to override the
	// client-level setting and ensure immediate locking during reads.
	txnOpts := spanner.TransactionOptions{
		ReadLockMode: pb.TransactionOptions_ReadWrite_PESSIMISTIC,
	}

	_, err = client.ReadWriteTransactionWithOptions(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
		// In PESSIMISTIC mode with SERIALIZABLE isolation, the transaction
		// acquires a shared lock during this read.
		key := spanner.Key{1, 2}
		row, err := txn.ReadRow(ctx, "Albums", key, []string{"AlbumTitle"})
		if err != nil {
			return fmt.Errorf("failed to read album: %w", err)
		}
		var title string
		if err := row.Column(0, &title); err != nil {
			return fmt.Errorf("failed to get album title: %w", err)
		}
		fmt.Fprintf(w, "Current album title: %s\n", title)

		// Update the album title
		stmt := spanner.Statement{
			SQL: `UPDATE Albums
				SET AlbumTitle = @AlbumTitle
				WHERE SingerId = @SingerId AND AlbumId = @AlbumId`,
			Params: map[string]interface{}{
				"SingerId":   1,
				"AlbumId":    2,
				"AlbumTitle": "New Album Title",
			},
		}
		count, err := txn.Update(ctx, stmt)
		if err != nil {
			return fmt.Errorf("failed to update album: %w", err)
		}
		fmt.Fprintf(w, "Updated %d record(s).\n", count)
		return nil
	}, txnOpts)

	if err != nil {
		return fmt.Errorf("transaction failed: %w", err)
	}
	return nil
}

Node.js

// Imports the Google Cloud Spanner client library
const {Spanner, protos} = require('@google-cloud/spanner');
// The read lock mode specified at the client-level will be applied
// to all RW transactions.
const defaultTransactionOptions = {
  readLockMode:
    protos.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode
      .OPTIMISTIC,
};

// Instantiates a client with defaultTransactionOptions
const spanner = new Spanner({
  projectId: projectId,
  defaultTransactionOptions,
});

function runTransactionWithReadLockMode() {
  // Gets a reference to a Cloud Spanner instance and database
  const instance = spanner.instance(instanceId);
  const database = instance.database(databaseId);
  // The read lock mode specified at the request-level takes precedence over
  // the read lock mode configured at the client-level.
  const readLockModeOptionsForTransaction = {
    readLockMode:
      protos.google.spanner.v1.TransactionOptions.ReadWrite.ReadLockMode
        .PESSIMISTIC,
  };

  database.runTransaction(
    readLockModeOptionsForTransaction,
    async (err, transaction) => {
      if (err) {
        console.error(err);
        return;
      }
      try {
        const query =
          'SELECT AlbumTitle FROM Albums WHERE SingerId = 2 AND AlbumId = 1';
        const results = await transaction.run(query);
        // Gets first album's title
        const rows = results[0].map(row => row.toJSON());
        const albumTitle = rows[0].AlbumTitle;
        console.log(`previous album title ${albumTitle}`);

        const update =
          "UPDATE Albums SET AlbumTitle = 'New Album Title' WHERE SingerId = 2 AND AlbumId = 1";
        const [rowCount] = await transaction.runUpdate(update);
        console.log(
          `Successfully updated ${rowCount} record in Albums table.`,
        );
        await transaction.commit();
        console.log(
          'Successfully executed read-write transaction with readLockMode option.',
        );
      } catch (err) {
        console.error('ERROR:', err);
        transaction.end();
      } finally {
        // Close the database when finished.
        await database.close();
      }
    },
  );
}
runTransactionWithReadLockMode();

Python

# instance_id = "your-spanner-instance"
# database_id = "your-spanner-db-id"
from google.cloud.spanner_v1 import TransactionOptions, DefaultTransactionOptions

# The read lock mode specified at the client-level will be applied to all
# RW transactions.
read_lock_mode_options_for_client = TransactionOptions.ReadWrite.ReadLockMode.OPTIMISTIC

# Create a client that uses Serializable isolation (default) with
# optimistic locking for read-write transactions.
spanner_client = spanner.Client(
    default_transaction_options=DefaultTransactionOptions(
        read_lock_mode=read_lock_mode_options_for_client
    )
)
instance = spanner_client.instance(instance_id)
database = instance.database(database_id)

# The read lock mode specified at the request level takes precedence over
# the read lock mode configured at the client level.
read_lock_mode_options_for_transaction = (
    TransactionOptions.ReadWrite.ReadLockMode.PESSIMISTIC
)

def update_albums_with_read_lock_mode(transaction):
    # Read an AlbumTitle.
    results = transaction.execute_sql(
        "SELECT AlbumTitle from Albums WHERE SingerId = 2 and AlbumId = 1"
    )
    for result in results:
        print("Current Album Title: {}".format(*result))

    # Update the AlbumTitle.
    row_ct = transaction.execute_update(
        "UPDATE Albums SET AlbumTitle = 'A New Title' WHERE SingerId = 2 and AlbumId = 1"
    )

    print("{} record(s) updated.".format(row_ct))

database.run_in_transaction(
    update_albums_with_read_lock_mode,
    read_lock_mode=read_lock_mode_options_for_transaction
)

C#

using Google.Cloud.Spanner.Data;
using System;
using System.Threading;
using System.Threading.Tasks;

public class ReadLockModeAsyncSample
{
    public async Task ReadLockModeAsync(string projectId, string instanceId, string databaseId)
    {
        // Create client with ReadLockMode.Optimistic.
        string connectionString = $"Data Source=projects/{projectId}/instances/{instanceId}/databases/{databaseId};ReadLockMode=Optimistic";

        using var connection = new SpannerConnection(connectionString);
        await connection.OpenAsync();

        // Create transaction options with ReadLockMode.Pessimistic.
        var transactionOptions = SpannerTransactionCreationOptions.ReadWrite
            .WithReadLockMode(ReadLockMode.Pessimistic);

        using var transaction = await connection.BeginTransactionAsync(transactionOptions, null, CancellationToken.None);

        var cmd = connection.CreateSelectCommand("SELECT AlbumTitle FROM Albums WHERE SingerId = 2 AND AlbumId = 1");
        cmd.Transaction = transaction;
        using (var reader = await cmd.ExecuteReaderAsync())
        {
            while (await reader.ReadAsync())
            {
                Console.WriteLine($"AlbumTitle: {reader.GetFieldValue<string>("AlbumTitle")}");
            }
        }

        var updateCmd = connection.CreateDmlCommand("UPDATE Albums SET AlbumTitle = 'A New Title' WHERE SingerId = 2 AND AlbumId = 1");
        updateCmd.Transaction = transaction;
        var rowCount = await updateCmd.ExecuteNonQueryAsync();
        Console.WriteLine($"{rowCount} records updated.");

        await transaction.CommitAsync();
    }
}

C++

void ReadLockModeSetting(std::string const& project_id,
                         std::string const& instance_id,
                         std::string const& database_id) {
  namespace spanner = ::google::cloud::spanner;
  using ::google::cloud::Options;
  using ::google::cloud::StatusOr;

  auto db = spanner::Database(project_id, instance_id, database_id);

  // The read lock mode specified at the client-level will be applied
  // to all RW transactions.
  auto options = Options{}.set<spanner::TransactionReadLockModeOption>(
      spanner::Transaction::ReadLockMode::kOptimistic);
  auto client = spanner::Client(spanner::MakeConnection(db, options));

  auto commit = client.Commit(
      [&client](
          spanner::Transaction const& txn) -> StatusOr<spanner::Mutations> {
        // Read an AlbumTitle.
        auto sql = spanner::SqlStatement(
            "SELECT AlbumTitle from Albums WHERE SingerId = @SingerId and "
            "AlbumId = @AlbumId",
            {{"SingerId", spanner::Value(2)}, {"AlbumId", spanner::Value(1)}});
        auto rows = client.ExecuteQuery(txn, std::move(sql));
        for (auto const& row :
             spanner::StreamOf<std::tuple<std::string>>(rows)) {
          if (!row) return row.status();
          std::cout << "Current Album Title: " << std::get<0>(*row) << "\n";
        }

        // Update the title.
        auto update_sql = spanner::SqlStatement(
            "UPDATE Albums "
            "SET AlbumTitle = @AlbumTitle "
            "WHERE SingerId = @SingerId and AlbumId = @AlbumId",
            {{"AlbumTitle", spanner::Value("A New Title")},
             {"SingerId", spanner::Value(2)},
             {"AlbumId", spanner::Value(1)}});
        auto result = client.ExecuteDml(txn, std::move(update_sql));
        if (!result) return result.status();
        std::cout << result->RowsModified() << " record(s) updated.\n";

        return spanner::Mutations{};
      },
      // The read lock mode specified at the transaction-level takes
      // precedence over the read lock mode configured at the client-level.
      // kPessimistic is used here to demonstrate overriding the client-level
      // setting.
      Options{}.set<spanner::TransactionReadLockModeOption>(
          spanner::Transaction::ReadLockMode::kPessimistic));

  if (!commit) throw std::move(commit).status();
  std::cout << "Update was successful [spanner_read_lock_mode]\n";
}

PHP

use Google\Cloud\Spanner\SpannerClient;
use Google\Cloud\Spanner\Transaction;
use Google\Cloud\Spanner\V1\TransactionOptions\ReadWrite\ReadLockMode;

/**
 * Shows how to run a Read Write transaction with read lock mode options.
 *
 * Example:
 * ```
 * read_lock_mode($instanceId, $databaseId);
 * ```
 *
 * @param string $instanceId The Spanner instance ID.
 * @param string $databaseId The Spanner database ID.
 */
function read_lock_mode(string $instanceId, string $databaseId): void
{
    // The read lock mode specified at the client-level will be applied to all
    // RW transactions.
    $spanner = new SpannerClient([
        'readLockMode' => ReadLockMode::OPTIMISTIC
    ]);
    $instance = $spanner->instance($instanceId);
    $database = $instance->database($databaseId);

    // The read lock mode specified at the request level takes precedence over
    // the read lock mode configured at the client level.
    $database->runTransaction(function (Transaction $t) {
        // Read an AlbumTitle.
        $results = $t->execute('SELECT AlbumTitle from Albums WHERE SingerId = 2 and AlbumId = 1');
        foreach ($results as $row) {
            printf('Current Album Title: %s' . PHP_EOL, $row['AlbumTitle']);
        }

        // Update the AlbumTitle.
        $rowCount = $t->executeUpdate('UPDATE Albums SET AlbumTitle = \'A New Title\' WHERE SingerId = 2 and AlbumId = 1');

        // Commit the transaction!
        $t->commit();

        printf('%d record(s) updated.' . PHP_EOL, $rowCount);
    }, [
        'transactionOptions' => [
            'readLockMode' => ReadLockMode::PESSIMISTIC
        ]
    ]);
}

Ruby

require "google/cloud/spanner"

def spanner_read_lock_mode project_id:, instance_id:, database_id:
  # Instantiates a client with read_lock_mode: :OPTIMISTIC
  spanner = Google::Cloud::Spanner.new project: project_id
  client = spanner.client instance_id, database_id, read_lock_mode: :OPTIMISTIC

  # Overrides read_lock_mode to :PESSIMISTIC at transaction level
  client.transaction read_lock_mode: :PESSIMISTIC do |tx|
    results = tx.execute_query "SELECT AlbumTitle FROM Albums WHERE SingerId = 2 AND AlbumId = 1"

    results.rows.each do |row|
      puts "AlbumTitle: #{row[:AlbumTitle]}"
    end

    row_count = tx.execute_update "UPDATE Albums SET AlbumTitle = 'A New Title' WHERE SingerId = 2 AND AlbumId = 1"

    puts "#{row_count} records updated."
  end
end

REST

Spanner TransactionOptions REST API 在 ReadWrite 消息中提供了一个 ReadLockMode 枚举，可让您选择 PESSIMISTIC 或 OPTIMISTIC 锁模式。

RPC

Spanner Transactionoptions RPC API 在 ReadWrite 消息中提供了一个 ReadLockMode 枚举，可让您选择 PESSIMISTIC 或 OPTIMISTIC 锁模式。

驱动程序

您可以使用 Spanner 的驱动程序将 read_lock_mode 设置为连接级别的连接参数，或设置为事务级别的 SET 语句选项。如需详细了解每个驱动程序，请参阅 驱动程序概览。

后续步骤

• 详细了解 Spanner 隔离级别。
• 了解如何使用可重复读隔离。