Aerospike高性能Go客户端使用指南

Aerospike是一个高性能、分布式的NoSQL数据库，而aerospike-client-go是其官方Go语言客户端库。下面我将详细介绍如何使用这个客户端库进行高效的数据操作。

1. 安装与初始化

首先安装客户端库：

go get github.com/aerospike/aerospike-client-go/v6

初始化客户端连接：

package main

import (
	"fmt"
	"log"
	"time"

	aerospike "github.com/aerospike/aerospike-client-go/v6"
)

func main() {
	// 配置客户端参数
	clientPolicy := aerospike.NewClientPolicy()
	clientPolicy.Timeout = 30 * time.Second // 连接超时时间
	clientPolicy.IdleTimeout = 60 * time.Second // 空闲连接超时

	// 创建客户端连接
	client, err := aerospike.NewClientWithPolicy(clientPolicy, "localhost", 3000)
	if err != nil {
		log.Fatalf("连接Aerospike失败: %v", err)
	}
	defer client.Close()

	// 检查集群状态
	if !client.IsConnected() {
		log.Fatal("客户端未连接到任何节点")
	}

	fmt.Println("成功连接到Aerospike集群")
}

2. 基本CRUD操作

写入数据

func writeRecord(client *aerospike.Client) {
	namespace := "test"
	setName := "users"
	key, err := aerospike.NewKey(namespace, setName, "user1")
	if err != nil {
		log.Fatal(err)
	}

	// 创建bin(列)
	bins := aerospike.BinMap{
		"username": "john_doe",
		"email":    "john@example.com",
		"age":      30,
		"active":   true,
	}

	// 写入策略
	writePolicy := aerospike.NewWritePolicy(0, 0)
	writePolicy.SendKey = true // 存储主键

	err = client.Put(writePolicy, key, bins)
	if err != nil {
		log.Fatalf("写入记录失败: %v", err)
	}

	fmt.Println("记录写入成功")
}

读取数据

func readRecord(client *aerospike.Client) {
	namespace := "test"
	setName := "users"
	key, err := aerospike.NewKey(namespace, setName, "user1")
	if err != nil {
		log.Fatal(err)
	}

	// 读取策略
	policy := aerospike.NewPolicy()
	policy.SendKey = true

	record, err := client.Get(policy, key)
	if err != nil {
		log.Fatalf("读取记录失败: %v", err)
	}

	if record == nil {
		fmt.Println("记录不存在")
		return
	}

	fmt.Printf("记录内容: %+v\n", record.Bins)
	fmt.Printf("记录的元数据: Generation=%d, TTL=%d\n", 
		record.Generation, record.Expiration)
}

更新数据

func updateRecord(client *aerospike.Client) {
	namespace := "test"
	setName := "users"
	key, err := aerospike.NewKey(namespace, setName, "user1")
	if err != nil {
		log.Fatal(err)
	}

	// 只更新age字段
	bins := aerospike.BinMap{
		"age": 31,
	}

	writePolicy := aerospike.NewWritePolicy(0, 0)
	writePolicy.SendKey = true

	err = client.Put(writePolicy, key, bins)
	if err != nil {
		log.Fatalf("更新记录失败: %v", err)
	}

	fmt.Println("记录更新成功")
}

删除数据

func deleteRecord(client *aerospike.Client) {
	namespace := "test"
	setName := "users"
	key, err := aerospike.NewKey(namespace, setName, "user1")
	if err != nil {
		log.Fatal(err)
	}

	// 删除策略
	deletePolicy := aerospike.NewWritePolicy(0, 0)
	existed, err := client.Delete(deletePolicy, key)
	if err != nil {
		log.Fatalf("删除记录失败: %v", err)
	}

	if existed {
		fmt.Println("记录删除成功")
	} else {
		fmt.Println("记录不存在")
	}
}

3. 批量操作

批量读取

func batchRead(client *aerospike.Client) {
	namespace := "test"
	setName := "users"

	// 创建多个key
	keys := []*aerospike.Key{}
	for i := 1; i <= 5; i++ {
		key, err := aerospike.NewKey(namespace, setName, fmt.Sprintf("user%d", i))
		if err != nil {
			log.Fatal(err)
		}
		keys = append(keys, key)
	}

	// 批量读取策略
	policy := aerospike.NewBatchPolicy()
	policy.SendKey = true

	records, err := client.BatchGet(policy, keys)
	if err != nil {
		log.Fatalf("批量读取失败: %v", err)
	}

	for i, record := range records {
		if record != nil {
			fmt.Printf("Key %d: %+v\n", i+1, record.Bins)
		} else {
			fmt.Printf("Key %d: 不存在\n", i+1)
		}
	}
}

批量写入

func batchWrite(client *aerospike.Client) {
	namespace := "test"
	setName := "users"

	writePolicy := aerospike.NewWritePolicy(0, 0)
	writePolicy.SendKey = true

	// 创建批量写入操作
	ops := []*aerospike.Operation{}
	for i := 1; i <= 5; i++ {
		key, err := aerospike.NewKey(namespace, setName, fmt.Sprintf("batch_user%d", i))
		if err != nil {
			log.Fatal(err)
		}

		bins := aerospike.BinMap{
			"username": fmt.Sprintf("user%d", i),
			"value":    i * 10,
		}

		err = client.Put(writePolicy, key, bins)
		if err != nil {
			log.Printf("写入记录 %d 失败: %v", i, err)
		}
	}

	fmt.Println("批量写入完成")
}

4. 查询与扫描

二级索引查询

func queryWithIndex(client *aerospike.Client) {
	namespace := "test"
	setName := "users"
	indexName := "age_index"
	binName := "age"

	// 创建二级索引(通常只需要执行一次)
	task, err := client.CreateIndex(
		nil, // 默认策略
		namespace, 
		setName, 
		indexName, 
		binName, 
		aerospike.NUMERIC,
	)
	if err != nil {
		log.Printf("创建索引可能已存在: %v", err)
	}

	// 等待索引创建完成
	if task != nil {
		<-task.OnComplete()
	}

	// 执行范围查询
	stmt := aerospike.NewStatement(namespace, setName)
	stmt.SetFilter(aerospike.NewRangeFilter(binName, 25, 35))

	recordset, err := client.Query(nil, stmt)
	if err != nil {
		log.Fatalf("查询失败: %v", err)
	}

	for res := range recordset.Results() {
		if res.Err != nil {
			log.Printf("处理记录时出错: %v", res.Err)
			continue
		}
		fmt.Printf("查询结果: %+v\n", res.Record.Bins)
	}
}

全表扫描

func scanAll(client *aerospike.Client) {
	namespace := "test"
	setName := "users"

	// 创建扫描策略
	policy := aerospike.NewScanPolicy()
	policy.ConcurrentNodes = true // 并行扫描节点
	policy.Priority = aerospike.HIGH
	policy.IncludeBinData = true

	recordset, err := client.ScanAll(policy, namespace, setName)
	if err != nil {
		log.Fatalf("扫描失败: %v", err)
	}

	count := 0
	for res := range recordset.Results() {
		if res.Err != nil {
			log.Printf("处理记录时出错: %v", res.Err)
			continue
		}
		count++
		if count <= 5 { // 只打印前5条记录
			fmt.Printf("扫描结果 %d: %+v\n", count, res.Record.Bins)
		}
	}

	fmt.Printf("总共扫描到 %d 条记录\n", count)
}

5. 性能优化建议

1. 连接池管理:

   • 客户端自动管理连接池，通过ClientPolicy调整池大小
   • 默认每个节点保持1个连接，可通过ClientPolicy.ConnectionQueueSize调整

2. 批量操作:

   • 尽量使用批量操作减少网络往返
   • 批量大小建议在100-1000之间，根据数据大小调整

3. 异步操作:

   • 使用ExecuteTask进行异步操作
   • 对于大量写入，考虑使用后台任务

4. 策略调优:

   • 根据场景选择合适的ConsistencyLevel
   • 对于不重要的数据可以使用ReplicaPolicy.MASTER提高性能

5. 压缩:

   • 对于大型BLOB数据启用压缩
   • 设置ClientPolicy.CompressionThreshold和压缩类型

6. 错误处理与重试

func robustOperation(client *aerospike.Client) {
	key, err := aerospike.NewKey("test", "users", "retry_user")
	if err != nil {
		log.Fatal(err)
	}

	// 配置重试策略
	policy := aerospike.NewPolicy()
	policy.MaxRetries = 3
	policy.SleepBetweenRetries = 200 * time.Millisecond

	bins := aerospike.BinMap{
		"data": "important information",
	}

	// 带重试的写入操作
	for i := 0; i <= policy.MaxRetries; i++ {
		err = client.Put(policy, key, bins)
		if err == nil {
			fmt.Println("写入成功")
			return
		}

		if i == policy.MaxRetries {
			log.Fatalf("操作失败，达到最大重试次数: %v", err)
		}

		log.Printf("尝试 %d 失败: %v, 重试中...", i+1, err)
		time.Sleep(policy.SleepBetweenRetries)
	}
}

通过以上示例，您应该能够掌握aerospike-client-go的基本使用方法。Aerospike客户端提供了丰富的功能和灵活的配置选项，可以根据具体应用场景进行调整以获得最佳性能。