Spanner客户端库 #

Spanner客户端库支持:
├── Java
├── Python
├── Go
├── Node.js
├── C++
├── C#
├── PHP
├── Ruby
└── REST API

客户端库功能:
├── 连接管理
├── 会话池
├── 事务处理
├── 查询执行
├── Mutation操作
├── 错误处理
└── 重试机制

<!-- Maven -->
<dependency>
    <groupId>com.google.cloud</groupId>
    <artifactId>google-cloud-spanner</artifactId>
    <version>6.45.0</version>
</dependency>

// Gradle
implementation 'com.google.cloud:google-cloud-spanner:6.45.0'

import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.spanner.Spanner;
import com.google.cloud.spanner.DatabaseId;
import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.Statement;
import com.google.cloud.spanner.ResultSet;

public class SpannerExample {
    
    public static void main(String[] args) {
        // 创建Spanner客户端
        SpannerOptions options = SpannerOptions.newBuilder().build();
        Spanner spanner = options.getService();
        
        // 获取数据库客户端
        DatabaseId dbId = DatabaseId.of("my-project", "my-instance", "my-database");
        DatabaseClient client = spanner.getDatabaseClient(dbId);
        
        // 执行查询
        try (ResultSet rs = client.singleUse()
                .executeQuery(Statement.of("SELECT * FROM users LIMIT 10"))) {
            while (rs.next()) {
                System.out.println(rs.getLong("user_id") + ": " + rs.getString("name"));
            }
        }
        
        spanner.close();
    }
}

import com.google.cloud.spanner.TransactionRunner;
import com.google.cloud.spanner.Mutation;
import com.google.cloud.spanner.Key;
import com.google.cloud.spanner.Struct;

// 读写事务
TransactionRunner runner = client.readWriteTransaction();
runner.run(transaction -> {
    // 读取数据
    Struct row = transaction.readRow("users", 
        Key.of(1L), Arrays.asList("name", "email"));
    
    // 更新数据
    transaction.buffer(Mutation.newUpdateBuilder("users")
        .set("user_id").to(1L)
        .set("name").to("John Updated")
        .build());
    
    return null;
});

// 只读事务
try (ReadOnlyTransaction transaction = client.readOnlyTransaction()) {
    ResultSet rs = transaction.executeQuery(
        Statement.of("SELECT * FROM users"));
    // 处理结果
}

import com.google.cloud.spanner.Mutation;

// 插入
Mutation insert = Mutation.newInsertBuilder("users")
    .set("user_id").to(1L)
    .set("name").to("John Doe")
    .set("email").to("john@example.com")
    .build();

// 更新
Mutation update = Mutation.newUpdateBuilder("users")
    .set("user_id").to(1L)
    .set("name").to("John Updated")
    .build();

// 删除
Mutation delete = Mutation.delete("users", Key.of(1L));

// 批量写入
List<Mutation> mutations = Arrays.asList(insert, update);
client.write(mutations);

pip install google-cloud-spanner

from google.cloud import spanner

# 创建客户端
spanner_client = spanner.Client(project='my-project')
instance = spanner_client.instance('my-instance')
database = instance.database('my-database')

# 执行查询
with database.snapshot() as snapshot:
    results = snapshot.execute_sql(
        "SELECT * FROM users LIMIT 10"
    )
    
    for row in results:
        print(f"{row[0]}: {row[1]}")

# 读写事务
def update_user(database, user_id, new_name):
    def transaction_func(transaction):
        # 读取数据
        row = list(transaction.read(
            "users", ["name", "email"],
            spanner.KeySet([spanner.Key([user_id])])
        ))[0]
        
        # 更新数据
        transaction.update(
            table="users",
            columns=("user_id", "name"),
            values=[(user_id, new_name)]
        )
    
    database.run_in_transaction(transaction_func)

# 只读事务
with database.snapshot() as snapshot:
    results = snapshot.execute_sql("SELECT * FROM users")

# 批量插入
def batch_insert(database, users):
    with database.batch() as batch:
        batch.insert(
            table="users",
            columns=("user_id", "name", "email"),
            values=users
        )

# 批量更新
def batch_update(database, users):
    with database.batch() as batch:
        batch.update(
            table="users",
            columns=("user_id", "name"),
            values=users
        )

# 批量删除
def batch_delete(database, user_ids):
    keys = [spanner.Key([uid]) for uid in user_ids]
    with database.batch() as batch:
        batch.delete("users", spanner.KeySet(keys=keys))

go get cloud.google.com/go/spanner

package main

import (
    "context"
    "fmt"
    "cloud.google.com/go/spanner"
)

func main() {
    ctx := context.Background()
    
    // 创建客户端
    dbPath := "projects/my-project/instances/my-instance/databases/my-database"
    client, err := spanner.NewClient(ctx, dbPath)
    if err != nil {
        panic(err)
    }
    defer client.Close()
    
    // 执行查询
    iter := client.Single().Query(ctx, spanner.Statement{
        SQL: "SELECT * FROM users LIMIT 10",
    })
    defer iter.Stop()
    
    for {
        row, err := iter.Next()
        if err == iterator.Done {
            break
        }
        if err != nil {
            panic(err)
        }
        
        var userID int64
        var name string
        if err := row.Columns(&userID, &name); err != nil {
            panic(err)
        }
        fmt.Printf("%d: %s\n", userID, name)
    }
}

// 读写事务
func updateUser(ctx context.Context, client *spanner.Client, userID int64, newName string) error {
    _, err := client.ReadWriteTransaction(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
        // 读取数据
        row, err := txn.ReadRow(ctx, "users", spanner.Key{userID}, []string{"name"})
        if err != nil {
            return err
        }
        
        // 更新数据
        m := spanner.Update("users", 
            []string{"user_id", "name"}, 
            []interface{}{userID, newName})
        
        return txn.BufferWrite([]*spanner.Mutation{m})
    })
    return err
}

// 只读事务
func readOnlyTransaction(ctx context.Context, client *spanner.Client) error {
    ro := client.ReadOnlyTransaction()
    defer ro.Close()
    
    iter := ro.Query(ctx, spanner.Statement{
        SQL: "SELECT * FROM users",
    })
    defer iter.Stop()
    
    // 处理结果...
    return nil
}

// 插入
func insertUser(ctx context.Context, client *spanner.Client, userID int64, name, email string) error {
    m := spanner.Insert("users",
        []string{"user_id", "name", "email"},
        []interface{}{userID, name, email},
    )
    
    _, err := client.Apply(ctx, []*spanner.Mutation{m})
    return err
}

// 批量操作
func batchInsert(ctx context.Context, client *spanner.Client, users []User) error {
    var mutations []*spanner.Mutation
    
    for _, u := range users {
        m := spanner.Insert("users",
            []string{"user_id", "name", "email"},
            []interface{}{u.ID, u.Name, u.Email},
        )
        mutations = append(mutations, m)
    }
    
    _, err := client.Apply(ctx, mutations)
    return err
}

npm install @google-cloud/spanner

const { Spanner } = require('@google-cloud/spanner');

// 创建客户端
const spanner = new Spanner({
    projectId: 'my-project',
});

const instance = spanner.instance('my-instance');
const database = instance.database('my-database');

// 执行查询
async function queryUsers() {
    const [rows] = await database.run({
        sql: 'SELECT * FROM users LIMIT 10',
    });
    
    rows.forEach(row => {
        console.log(`${row.user_id}: ${row.name}`);
    });
}

queryUsers();

// 读写事务
async function updateUser(userId, newName) {
    await database.runTransactionAsync(async (transaction) => {
        // 读取数据
        const [rows] = await transaction.run({
            sql: 'SELECT name FROM users WHERE user_id = @userId',
            params: { userId: userId },
        });
        
        // 更新数据
        await transaction.runUpdate({
            sql: 'UPDATE users SET name = @name WHERE user_id = @userId',
            params: { name: newName, userId: userId },
        });
        
        await transaction.commit();
    });
}

// 只读事务
async function readOnlyTransaction() {
    const [rows] = await database.run({
        sql: 'SELECT * FROM users',
    });
    // 处理结果
}

// 批量插入
async function batchInsert(users) {
    const mutations = users.map(user => ({
        insert: {
            table: 'users',
            columns: ['user_id', 'name', 'email'],
            values: [[user.id.toString(), user.name, user.email]],
        },
    }));
    
    await database.write(mutations);
}

import com.google.cloud.spanner.SpannerOptions;
import com.google.cloud.spanner.SessionPoolOptions;

SessionPoolOptions poolOptions = SessionPoolOptions.newBuilder()
    .setMinSessions(10)
    .setMaxSessions(100)
    .setIncStep(5)
    .build();

SpannerOptions options = SpannerOptions.newBuilder()
    .setSessionPoolOption(poolOptions)
    .build();

from google.cloud import spanner

client = spanner.Client(
    project='my-project',
    client_options={
        'session_pool_options': {
            'min_sessions': 10,
            'max_sessions': 100,
        }
    }
)

import "cloud.google.com/go/spanner"

config := spanner.ClientConfig{
    SessionPoolConfig: spanner.SessionPoolConfig{
        MinOpened: 10,
        MaxOpened: 100,
    },
}
client, err := spanner.NewClientWithConfig(ctx, dbPath, config)

import com.google.cloud.spanner.SpannerException;
import com.google.cloud.spanner.ErrorCode;

try {
    client.write(mutations);
} catch (SpannerException e) {
    switch (e.getErrorCode()) {
        case ALREADY_EXISTS:
            System.err.println("主键冲突");
            break;
        case NOT_FOUND:
            System.err.println("数据不存在");
            break;
        case DEADLINE_EXCEEDED:
            System.err.println("操作超时");
            break;
        case ABORTED:
            System.err.println("事务被中止");
            break;
        default:
            System.err.println("其他错误: " + e.getMessage());
    }
}

from google.cloud.spanner import exceptions

try:
    database.run_in_transaction(transaction_func)
except exceptions.AlreadyExists:
    print("主键冲突")
except exceptions.NotFound:
    print("数据不存在")
except exceptions.DeadlineExceeded:
    print("操作超时")
except exceptions.Aborted:
    print("事务被中止")

连接管理建议:
├── 使用连接池
├── 合理设置池大小
├── 及时关闭连接
├── 复用客户端实例
└── 监控连接状态

事务处理建议:
├── 减少事务大小
├── 使用只读事务
├── 处理冲突重试
├── 设置合理超时
└── 监控事务性能

性能优化建议:
├── 使用批量操作
├── 使用参数化查询
├── 避免热点主键
├── 使用交错表
└── 监控客户端性能

1. 使用连接池
   └── 合理配置池大小

2. 使用参数化查询
   └── 防止SQL注入

3. 处理错误和重试
   └── 保证可靠性

4. 监控客户端性能
   └── 及时发现问题

5. 复用客户端实例
   └── 减少创建开销