package clientv3
import (
pb "github.com/coreos/etcd/etcdserver/etcdserverpb" //protobuffer
"golang.org/x/net/context"
"google.golang.org/grpc" //google rpc 框架
)
type (
CompactResponse pb.CompactionResponse //带压缩的响应
PutResponse pb.PutResponse //添加响应
GetResponse pb.RangeResponse // 带区间的响应
DeleteResponse pb.DeleteRangeResponse // 删除带区间数据的响应
TxnResponse pb.TxnResponse //带事务的响应
)
type KV interface {
// Put puts a key-value pair into etcd.
//添加键值对到etcd中
// Note that key,value can be plain bytes array and string is
// an immutable representation of that bytes array.
//注意：键值对可以是字节数组或者字符串 字符串是原子性的字节数组
// To get a string of bytes, do string([]byte(0x10, 0x20)).
//
Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error)
// Get retrieves keys.
//获取键对应的值
// By default, Get will return the value for "key", if any.
//默认 获取键对应的值 在任何情况下
// When passed WithRange(end), Get will return the keys in the range [key, end).
//当opts 使用了 WithRange(end)，将得到键对应的区间[key, end)之间的值
// When passed WithFromKey(), Get returns keys greater than or equal to key.
//opts 使用了WithFromKey()，得到大于等于当前key 对应的value
// When passed WithRev(rev) with rev > 0, Get retrieves keys at the given revision;
//当opts 使用了 WithRev(rev)，如果版本号rev>0 获取指定的版本号对应的value
// if the required revision is compacted, the request will fail with ErrCompacted .
// When passed WithLimit(limit), the number of returned keys is bounded by limit.
//当opts 使用了 WithLimit(limit)，将得到键对应的区间[key最小值 默认的 end [limit )的值 例如： key为 foo 实际的key为 fooN。。。。到foolimit 之间对应的value
// When passed WithSort(), the keys will be sorted.
//当opts 使用了 WithSort()，得到的值将排序 按照字典的顺序来排序
Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error)
// Delete deletes a key, or optionally using WithRange(end), [key, end).
// 删除一个键值对 更常使用 WithRange(end), [key, end).
Delete(ctx context.Context, key string, opts ...OpOption) (*DeleteResponse, error)
// Compact compacts etcd KV history before the given rev.
// 压缩etcd kv历史数据 通常再给出版本号之前
Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error)
// Do applies a single Op on KV without a transaction.
//do应用于单一kv操作项，通常是没有事务的
// Do is useful when declaring operations to be issued at a later time
//
// whereas Get/Put/Delete are for better suited for when the operation
// should be immediately issued at time of declaration.
// Do applies a single Op on KV without a transaction.
// Do is useful when creating arbitrary operations to be issued at a
// later time; the user can range over the operations, calling Do to
// execute them. Get/Put/Delete, on the other hand, are best suited
// for when the operation should be issued at the time of declaration.
Do(ctx context.Context, op Op) (OpResponse, error)
// Txn creates a transaction.
//创建事务
Txn(ctx context.Context) Txn
}
//响应结构体
type OpResponse struct {
put *PutResponse
get *GetResponse
del *DeleteResponse
}
func (op OpResponse) Put() *PutResponse { return op.put }
func (op OpResponse) Get() *GetResponse { return op.get }
func (op OpResponse) Del() *DeleteResponse { return op.del }
//kv存储服务客户端的包装
type kv struct {
remote pb.KVClient
}
//创建kv 服务 带着指定的客户端
func NewKV(c *Client) KV {
return &kv{remote: RetryKVClient(c)}
}
//创建一个kv服务客户端 带着指定的protobuffer 客户端
func NewKVFromKVClient(remote pb.KVClient) KV {
return &kv{remote: remote}
}
//kv结构体实现了 kv接口
func (kv *kv) Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error) {
r, err := kv.Do(ctx, OpPut(key, val, opts...))
return r.put, toErr(ctx, err)
}
func (kv *kv) Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error) {
r, err := kv.Do(ctx, OpGet(key, opts...))
return r.get, toErr(ctx, err)
}
func (kv *kv) Delete(ctx context.Context, key string, opts ...OpOption) (*DeleteResponse, error) {
r, err := kv.Do(ctx, OpDelete(key, opts...))
return r.del, toErr(ctx, err)
}
func (kv *kv) Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error) {
resp, err := kv.remote.Compact(ctx, OpCompact(rev, opts...).toRequest())
if err != nil {
return nil, toErr(ctx, err)
}
return (*CompactResponse)(resp), err
}
func (kv *kv) Txn(ctx context.Context) Txn {
return &txn{
kv: kv,
ctx: ctx,
}
}
func (kv *kv) Do(ctx context.Context, op Op) (OpResponse, error) {
for {
resp, err := kv.do(ctx, op)
if err == nil {
return resp, nil
}
if isHaltErr(ctx, err) {
return resp, toErr(ctx, err)
}
// do not retry on modifications
if op.isWrite() {
return resp, toErr(ctx, err)
}
}
}
func (kv *kv) do(ctx context.Context, op Op) (OpResponse, error) {
var err error
switch op.t {
// TODO: handle other ops
case tRange:
var resp *pb.RangeResponse
resp, err = kv.remote.Range(ctx, op.toRangeRequest(), grpc.FailFast(false))
if err == nil {
return OpResponse{get: (*GetResponse)(resp)}, nil
}
case tPut:
var resp *pb.PutResponse
r := &pb.PutRequest{Key: op.key, Value: op.val, Lease: int64(op.leaseID), PrevKv: op.prevKV}
resp, err = kv.remote.Put(ctx, r)
if err == nil {
return OpResponse{put: (*PutResponse)(resp)}, nil
}
case tDeleteRange:
var resp *pb.DeleteRangeResponse
r := &pb.DeleteRangeRequest{Key: op.key, RangeEnd: op.end, PrevKv: op.prevKV}
resp, err = kv.remote.DeleteRange(ctx, r)
if err == nil {
return OpResponse{del: (*DeleteResponse)(resp)}, nil
}
default:
panic("Unknown op")
}
return OpResponse{}, err
}
时间： 2024-10-13 19:04:54