Welcome to Jayj's Blog

type kv struct {
    remote pb.KVClient
    callOpts []grpc.CallOption
}

cli, err := clientv3.New(clientv3.Config{
    Endpoints: "example.endpoint",
    DialTimeout: 5 * time.Second
})
if err := nil {
    return err
}
defer cli.Close()

func NewKV(c *Client) KV {
    api := &kv{remote: RetryKVClient(c)}
    if c != nil {
        api.callOpts = c.callOpts
    }

    return api
}

func (kv *kv) Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error)

kv := clientv3.NewKV(cli)

resp, err := kv.Put(context.Background(), key, val, ...opts)

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)
}

Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error)

resp, err := cli.Put(context.Background(), key, ...opts)

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)
}

Delete(ctx context.Context, key string, opts ...OpOption) (*DeleteResponse, error)

resp, err = cli.Delete(ctx, key, ...opts)

r, err := kv.Do(ctx, OpDelete(key, opts...))
return r.del, toErr(ctx, err)

Do(ctx context.Context, op Op) (OpResponse, error)

ops := []clientv3.Op{
   clientv3.OpPut("put-key", "123"),
   clientv3.OpGet("put-key"),
   clientv3.OpPut("put-key", "456")}

  for _, op := range ops {
   if _, err := cli.Do(context.TODO(), op); err != nil {
    log.Fatal(err)
   }
  }

func (kv *kv) Do(ctx context.Context, op Op) (OpResponse, error) {
 var err error
 switch op.t { // 对传入的op遍历
 case tRange:
  var resp *pb.RangeResponse
  resp, err = kv.remote.Range(ctx, op.toRangeRequest(), kv.callOpts...)
  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, IgnoreValue: op.ignoreValue, IgnoreLease: op.ignoreLease}
  resp, err = kv.remote.Put(ctx, r, kv.callOpts...)
  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, kv.callOpts...)
  if err == nil {
   return OpResponse{del: (*DeleteResponse)(resp)}, nil
  }
 case tTxn:
  var resp *pb.TxnResponse
  resp, err = kv.remote.Txn(ctx, op.toTxnRequest(), kv.callOpts...)
  if err == nil {
   return OpResponse{txn: (*TxnResponse)(resp)}, nil
  }
 default:
  panic("Unknown op")
 }
 return OpResponse{}, toErr(ctx, err)
}

Txn(ctx context.Context) Txn

kvc := clientv3.NewKV(cli)

  _, err = kvc.Put(context.TODO(), "key", "xyz")
  if err != nil {
   log.Fatal(err)
  }

  ctx, cancel := context.WithTimeout(context.Background(), requestTimeout)
  _, err = kvc.Txn(ctx).
   // 事务的值是字典类型的比较
   If(clientv3.Compare(clientv3.Value("key"), ">", "abc")).
   // "xyz" > "abc" 时运行
   Then(clientv3.OpPut("key", "XYZ")).
   // else运行
   Else(clientv3.OpPut("key", "ABC")).
   Commit()

func (kv *kv) Txn(ctx context.Context) Txn {
 return &txn{
  kv:       kv,
  ctx:      ctx,
  callOpts: kv.callOpts,
 }
}

Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error)

resp, err := cli.Get(ctx, "foo")
  cancel()
  if err != nil {
   log.Fatal(err)
  }
  compRev := resp.Header.Revision // 具体你要压缩哪个版本之前的key-val

  ctx, cancel = context.WithTimeout(context.Background(), requestTimeout)
  _, err = cli.Compact(ctx, compRev)

func (kv *kv) Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error) {
 resp, err := kv.remote.Compact(ctx, OpCompact(rev, opts...).toRequest(), kv.callOpts...)
 if err != nil {
  return nil, toErr(ctx, err)
 }
 return (*CompactResponse)(resp), err
}

type PingServer interface {
 Pong(context.Context, *Req) (*Resp, error)
}

package service

import (
 "context"
 "ping/proto/pb"
)

type PingService struct {
}

func NewPingService() *PingService {
 return &PingService{}
}

func (s *PingService) Pong(ctx context.Context, in *pb.Req) (*pb.Resp, error) {

 msg := fmt.Sprintf("hello %s", in.Name)

 return &pb.Resp{
  Msg: msg,
 }, nil
}

package main

import (
 "fmt"
 "net"
 "ping/discovery"
 "ping/proto/pb"
 "ping/service"

 "github.com/sirupsen/logrus"
 "google.golang.org/grpc"
)

const (
  // 微服务名称
 app         = "ping"
 // 微服务地址
 grpcAddress = "127.0.0.1:10004"
)

func main() {
 // etcd 地址
 etcdAddress := []string{"127.0.0.1:2379"}

 // 服务注册
 etcdRegister := discovery.NewRegister(etcdAddress, logrus.New())
 defer etcdRegister.Stop()

 node := discovery.Server{
  Name: app,
  Addr: grpcAddress,
 }

 server := grpc.NewServer()
 defer server.Stop()

 // 绑定service
 pb.RegisterPingServer(server, service.NewPingService())

 lis, err := net.Listen("tcp", grpcAddress)
 if err != nil {
  panic(err)
 }
 defer lis.Close()

 if _, err := etcdRegister.Register(node, 10); err != nil {
  panic(fmt.Sprintf("start server failed, err: %v", err))
 }

 logrus.Info("server started listen on ", grpcAddress)
 if err := server.Serve(lis); err != nil {
  panic(err)
 }
}

package routes

import (
 "gateway/proto/pb"
 "gateway/utils/res"

 "github.com/gin-gonic/gin"
 "github.com/sirupsen/logrus"
 "google.golang.org/grpc"
 "google.golang.org/grpc/balancer/roundrobin"
)

func addPingRoute(rg *gin.RouterGroup) {
 opts := []grpc.DialOption{
  grpc.WithInsecure(),
  // 在这里出现了负载均衡，在完成了整个项目之后你可以创建两个ping服务同时注册
  // 同时通过网关请求多次试一试
  // 你会发现 一般情况下，两个ping服务会平分请求，也就达到了负载均衡
  // 具体均衡规则还可以通过配置weight来实现
  grpc.WithBalancerName(roundrobin.Name),
 }

 // 通过etcd做服务发现 在这里你并不是通过请求ping服务的10004来连接而是通过etcd去寻找ping服务
 pingConn, err := grpc.Dial("etcd:///ping", opts...)
 if err != nil {
  logrus.Errorf("try connect ping service failed")
 }
 // 不要调用pingConn.Close() 因为addRoute方法只是导入了路由与对应的方法
 // 在读取完毕之后会关闭链接 读取完毕之后的请求会无法链接 因为链接已经关闭
 // 如果你一定要严谨的关闭 就把它移出去到公共的地方注册链接 并创建方法在优雅关闭内结束

 pingClient := pb.NewPingClient(pingConn)

 // 在这里 分组 路由的方法等 如果用过单体gin的同学是不是很眼熟了
 ping := rg.Group("/ping")
 {
  ping.GET("", func(ctx *gin.Context) {
   name := ctx.Query("name")

   req := &pb.Req{
    Name: name,
   }

   resp, err := pingClient.Pong(ctx, req)
   if err != nil {
    logrus.Errorln("/v1/ping err, err: ", err)
    res.InternalError(ctx)
    return
   }

   // 这里是自己封装的方法
   res.Ok(ctx, res.OK, resp.Msg)
  })
 }
}

package routes

import (
 "gateway/middleware"
 "gateway/utils/res"
 "net/http"

 "github.com/gin-gonic/gin"
)

// LoadGin 初始化gin 注册路由
func LoadGin() *gin.Engine {

 g := gin.Default()

 g.Use(middleware.Cors())

 getRoute(g)

 g.NoRoute(func(ctx *gin.Context) {
  res.Error(ctx, http.StatusNotFound, res.UrlNotFound)
 })

 return g
}

func getRoute(g *gin.Engine) {

 v1 := g.Group("/v1")
 addPingRoute(v1)

  // ...
}

package main

import (
 "gateway/discovery"
 "gateway/routes"
 "gateway/utils"
 "net/http"
 "time"

 "github.com/sirupsen/logrus"
 "google.golang.org/grpc/resolver"
)

func main() {

 // etcd注册
 r := discovery.NewResolver([]string{"localhost:2379"}, logrus.New())
 defer r.Close()
 resolver.Register(r)

 // 初始化gin 加载路由
 g := routes.LoadGin()

 server := &http.Server{
  Addr:           ":8080",
  Handler:        g,
  ReadTimeout:    10 * time.Second,
  WriteTimeout:   10 * time.Second,
  MaxHeaderBytes: 1 << 20,
 }

 go func() {
  // 优雅关闭http server 源码内有该方法，你也可以直接抹掉这一行
  // 为什么要优雅关闭server？
  // https://studygolang.com/articles/12600
  utils.GracefullyShutdown(server)
 }()

 logrus.Info("gateway listen on :8080")

 if err := server.ListenAndServe(); err != nil {
  logrus.Fatal("gateway启动失败, err: ", err)
 }
}

package middleware

import (
 "gateway/utils"
 "gateway/utils/res"
 "time"

 "github.com/gin-gonic/gin"
)

// JWT token的生成是使用用户名、密码等一系列信息生成 在解密的时候可以拿到这些信息
func JWT() gin.HandlerFunc {
 return func(ctx *gin.Context) {
  var claims *utils.Claims
  var err error

  code := res.OK

  // 对于我个人喜欢将鉴权的token放到header中的Authorization中
  // 如果你有自己的想法，可以在这里获取你将token放的位置
  token := ctx.GetHeader("Authorization")
  if token == "" {
   res.Unauthorized(ctx, res.TokenInvalid)

   ctx.Abort()
   return
  }

  // 验证
  claims, err = utils.ParseToken(token)
  if err != nil {

   res.Unauthorized(ctx, res.TokenInvalid)

   ctx.Abort()
   return
  }

  // 判断是否过期
  if time.Now().Unix() > claims.ExpiresAt {
   code = res.TokenExpired
  } else if err != nil {
   code = res.TokenInvalid
  }

  if code == res.TokenExpired {
   res.Ok__(ctx, res.TokenExpired)
  } else if code != res.OK {

   res.Unauthorized(ctx, code)

   ctx.Abort()
   return
  }

  // 我在这里的操作是将token对应的user放在了上下文内，
  // 对于需要获取用户的方法，只通过这里就可以拿到userid
  // 而不再需要通过传输参数的方式传入用户信息
  ctx.Set("userid", claims.Userid)
  ctx.Set("username", claims.Username)
  ctx.Next()
 }
}

package routes

import (
 "gateway/middleware"
 "gateway/proto/pb"
 "gateway/utils/res"

 "github.com/gin-gonic/gin"
 "github.com/sirupsen/logrus"
 "google.golang.org/grpc"
 "google.golang.org/grpc/balancer/roundrobin"
)

func addPingRoute(rg *gin.RouterGroup) {
 opts := []grpc.DialOption{
  grpc.WithInsecure(),
  grpc.WithBalancerName(roundrobin.Name),
 }

 // 通过etcd做服务发现
 pingConn, err := grpc.Dial("etcd:///ping", opts...)
 if err != nil {
  logrus.Errorf("try connect ping service failed")
 }
 // 不要调用pingConn.Close() 因为addRoute方法只是导入了路由与对应的方法
 // 在读取完毕之后会关闭链接 在读取完毕之后的请求会无法链接 因为链接已经关闭
 // 如果你一定要严谨的关闭 就把它移出去到公共的地方注册链接 并创建方法在优雅关闭内结束链接

 pingClient := pb.NewPingClient(pingConn)

 ping := rg.Group("/ping")
 // 在这里使用中间件
 ping.Use(middleware.JWT())
 {
  ping.GET("", func(ctx *gin.Context) {
   name := ctx.Query("name")

   req := &pb.Req{
    Name: name,
   }

   resp, err := pingClient.Pong(ctx, req)
   if err != nil {
    logrus.Errorln("/v1/ping err, err: ", err)
    res.InternalError(ctx)
    return
   }

   res.Ok(ctx, res.OK, resp.Msg)
  })
 }
}