GoCouchbase (gocb) 使用指南

gocb 是 Couchbase 官方提供的 Go 语言 SDK，用于与 Couchbase Server 进行交互。下面我将详细介绍如何使用 gocb 进行基本的数据库操作。

安装

首先需要安装 gocb 包：

go get github.com/couchbase/gocb/v2

基本使用

1. 连接集群

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/couchbase/gocb/v2"
)

func main() {
	// 连接集群
	cluster, err := gocb.Connect(
		"couchbase://localhost",
		gocb.ClusterOptions{
			Username: "Administrator",
			Password: "password",
		})
	if err != nil {
		panic(err)
	}

	// 确保关闭连接
	defer cluster.Close(nil)

	// 获取bucket
	bucket := cluster.Bucket("travel-sample")

	// 等待bucket连接
	err = bucket.WaitUntilReady(5*time.Second, nil)
	if err != nil {
		panic(err)
	}

	// 获取集合(scope.collection)
	collection := bucket.Scope("inventory").Collection("airline")
	
	// 后续操作...
}

2. CRUD 操作

插入文档

type Airline struct {
	ID       string `json:"id"`
	Type     string `json:"type"`
	Name     string `json:"name"`
	Iata     string `json:"iata"`
	Icao     string `json:"icao"`
	Callsign string `json:"callsign"`
	Country  string `json:"country"`
}

// 插入文档
func insertDoc(collection *gocb.Collection) {
	airline := Airline{
		ID:       "airline_123",
		Type:     "airline",
		Name:     "Oceanic Airlines",
		Iata:     "OA",
		Icao:     "OAL",
		Callsign: "Oceanic",
		Country:  "United States",
	}

	// 插入文档，设置过期时间为1小时
	_, err := collection.Insert("airline_123", airline, &gocb.InsertOptions{
		Timeout: 5 * time.Second,
		Expiry:  1 * time.Hour,
	})
	if err != nil {
		fmt.Printf("插入失败: %v\n", err)
		return
	}
	fmt.Println("文档插入成功")
}

获取文档

// 获取文档
func getDoc(collection *gocb.Collection) {
	var airline Airline
	result, err := collection.Get("airline_123", &gocb.GetOptions{})
	if err != nil {
		fmt.Printf("获取失败: %v\n", err)
		return
	}

	err = result.Content(&airline)
	if err != nil {
		fmt.Printf("解析失败: %v\n", err)
		return
	}

	fmt.Printf("获取到的文档: %+v\n", airline)
}

更新文档

// 更新文档
func updateDoc(collection *gocb.Collection) {
	// 先获取当前文档
	var airline Airline
	result, err := collection.Get("airline_123", &gocb.GetOptions{})
	if err != nil {
		fmt.Printf("获取文档失败: %v\n", err)
		return
	}

	err = result.Content(&airline)
	if err != nil {
		fmt.Printf("解析文档失败: %v\n", err)
		return
	}

	// 修改文档
	airline.Country = "Canada"

	// 替换文档
	_, err = collection.Replace("airline_123", airline, &gocb.ReplaceOptions{
		Cas: result.Cas(),
	})
	if err != nil {
		fmt.Printf("更新失败: %v\n", err)
		return
	}
	fmt.Println("文档更新成功")
}

删除文档

// 删除文档
func deleteDoc(collection *gocb.Collection) {
	_, err := collection.Remove("airline_123", &gocb.RemoveOptions{})
	if err != nil {
		fmt.Printf("删除失败: %v\n", err)
		return
	}
	fmt.Println("文档删除成功")
}

3. 查询操作

N1QL 查询

// N1QL查询
func queryDocs(cluster *gocb.Cluster) {
	query := "SELECT * FROM `travel-sample`.inventory.airline WHERE country = $1 LIMIT 5"
	rows, err := cluster.Query(query, &gocb.QueryOptions{
		PositionalParameters: []interface{}{"United States"},
		Adhoc:                true,
	})
	if err != nil {
		fmt.Printf("查询失败: %v\n", err)
		return
	}

	var airlines []Airline
	for rows.Next() {
		var airline Airline
		err := rows.Row(&airline)
		if err != nil {
			fmt.Printf("解析行失败: %v\n", err)
			continue
		}
		airlines = append(airlines, airline)
	}

	// 检查查询元数据
	metadata, err := rows.MetaData()
	if err != nil {
		fmt.Printf("获取元数据失败: %v\n", err)
	}

	fmt.Printf("查询到 %d 条结果\n", len(airlines))
	fmt.Printf("执行时间: %v\n", metadata.Metrics.ExecutionTime)
	for _, a := range airlines {
		fmt.Printf("- %s (%s)\n", a.Name, a.Iata)
	}
}

范围查询

// 范围查询
func rangeQuery(collection *gocb.Collection) {
	query := gocb.NewRangeScan(
		gocb.ScanTerm{Term: "airline_100"},
		gocb.ScanTerm{Term: "airline_200"},
	)
	
	scanResult, err := collection.Scan(query, &gocb.ScanOptions{})
	if err != nil {
		fmt.Printf("范围查询失败: %v\n", err)
		return
	}

	var count int
	for scanResult.Next() {
		var doc Airline
		err := scanResult.Content(&doc)
		if err != nil {
			fmt.Printf("解析文档失败: %v\n", err)
			continue
		}
		count++
		fmt.Printf("文档ID: %s, 名称: %s\n", doc.ID, doc.Name)
	}

	fmt.Printf("共查询到 %d 个文档\n", count)
}

4. 批量操作

// 批量操作
func bulkOps(collection *gocb.Collection) {
	// 创建批量操作对象
	ops := []gocb.BulkOp{
		&gocb.GetOp{ID: "airline_123"},
		&gocb.UpsertOp{
			ID:     "airline_456",
			Value:  Airline{ID: "airline_456", Name: "New Airline", Country: "UK"},
			Expiry: 2 * time.Hour,
		},
		&gocb.RemoveOp{ID: "airline_789"},
	}

	// 执行批量操作
	err := collection.Do(ops, &gocb.BulkOpOptions{})
	if err != nil {
		fmt.Printf("批量操作失败: %v\n", err)
		return
	}

	// 处理结果
	for _, op := range ops {
		switch o := op.(type) {
		case *gocb.GetOp:
			if o.Err != nil {
				fmt.Printf("获取文档 %s 失败: %v\n", o.ID, o.Err)
				continue
			}
			var doc Airline
			err := o.Result.Content(&doc)
			if err != nil {
				fmt.Printf("解析文档 %s 失败: %v\n", o.ID, err)
				continue
			}
			fmt.Printf("获取到文档: %+v\n", doc)
		case *gocb.UpsertOp:
			if o.Err != nil {
				fmt.Printf("插入/更新文档 %s 失败: %v\n", o.ID, o.Err)
				continue
			}
			fmt.Printf("文档 %s 操作成功\n", o.ID)
		case *gocb.RemoveOp:
			if o.Err != nil {
				fmt.Printf("删除文档 %s 失败: %v\n", o.ID, o.Err)
				continue
			}
			fmt.Printf("文档 %s 删除成功\n", o.ID)
		}
	}
}

高级特性

1. 事务支持

// 事务操作
func transactionExample(cluster *gocb.Cluster, collection *gocb.Collection) {
	// 创建事务配置
	config := &gocb.TransactionsConfig{
		DurabilityLevel: gocb.DurabilityLevelPersistToMajority,
	}

	// 执行事务
	_, err := cluster.Transactions().Run(func(ctx *gocb.TransactionAttemptContext) error {
		// 获取文档
		getResult, err := ctx.Get(collection, "airline_123")
		if err != nil {
			return err
		}

		var doc Airline
		err = getResult.Content(&doc)
		if err != nil {
			return err
		}

		// 修改文档
		doc.Country = "Australia"

		// 替换文档
		_, err = ctx.Replace(getResult, doc)
		if err != nil {
			return err
		}

		// 插入新文档
		newDoc := Airline{
			ID:      "airline_999",
			Name:    "Transaction Airline",
			Country: "Australia",
		}
		_, err = ctx.Insert(collection, "airline_999", newDoc)
		if err != nil {
			return err
		}

		return nil
	}, config)

	if err != nil {
		fmt.Printf("事务执行失败: %v\n", err)
		return
	}

	fmt.Println("事务执行成功")
}

2. 订阅变更

// 订阅变更
func watchChanges(cluster *gocb.Cluster, collection *gocb.Collection) {
	// 创建变更订阅
	watcher, err := collection.Watch(
		[]gocb.WatchCreateOptions{
			{
				Key: "airline_123",
			},
		},
		&gocb.WatchOptions{
			Timeout: 30 * time.Second,
		},
	)
	if err != nil {
		fmt.Printf("创建watcher失败: %v\n", err)
		return
	}

	// 处理变更事件
	for {
		select {
		case event := <-watcher:
			switch event.EventType {
			case gocb.WatchEventTypeMutation:
				fmt.Printf("文档 %s 被修改\n", event.Key)
				var doc Airline
				err := event.Content(&doc)
				if err != nil {
					fmt.Printf("解析文档失败: %v\n", err)
					continue
				}
				fmt.Printf("新内容: %+v\n", doc)
			case gocb.WatchEventTypeDeletion:
				fmt.Printf("文档 %s 被删除\n", event.Key)
			case gocb.WatchEventTypeExpiration:
				fmt.Printf("文档 %s 已过期\n", event.Key)
			}
		case <-time.After(30 * time.Second):
			fmt.Println("监听超时")
			return
		}
	}
}

最佳实践

1. 连接池管理: 重用 Cluster 对象，避免频繁创建和销毁连接
2. 错误处理: 检查所有操作的错误返回
3. 超时设置: 为所有操作设置合理的超时时间
4. CAS 操作: 使用 CAS 值处理并发更新
5. 批量操作: 对于大量操作使用批量接口提高性能
6. 索引优化: 为常用查询创建合适的索引

以上是 gocb 的基本使用方法，更多高级功能可以参考官方文档：Couchbase Go SDK Documentation