命令记不住

etcd服务案例

1.Project总体

项目文件结构

$grpcdemo-etcd
.
├── client
│   ├── go.mod
│   ├── go.sum
│   └── main.go
├── etcdlock
│   └── main.go
├── etcdservice
│   ├── go.mod
│   ├── go.sum
│   ├── naming.go
│   └── resolver.go
├── protoes
│   ├── go.mod
│   ├── go.sum
│   ├── hello.pb.go
│   └── hello.proto
├── readme.txt
└── server
    ├── go.mod
    ├── go.sum
    └── main.go

2.Golang实战

启动三个Server

server.exe -Port 3000

server.exe -Port 3001

server.exe -Port 3002

checklist

1.启动客户端
    代码
    运行
2.观察服务器窗口
    三个服务轮流收到消息，负载均衡
    停止其中的服务，并重启，自动重新均衡访问
    停止所有服务，客户端退出
3.etcd重启，不中断客户端访问服务端

3.锲约protoes

go.mod

module protoes

go 1.14

replace github.com/coreos/go-systemd => github.com/coreos/go-systemd/v22 v22.0.0

require (
	github.com/golang/protobuf v1.4.2
	google.golang.org/grpc v1.29.1
)

hello.proto

syntax = "proto3";

package protoes;

service HelloServer{
    rpc SayHi(HelloRequest)returns(HelloReplay){}
    rpc GetMsg(HelloRequest)returns(HelloMessage){}
}

message HelloRequest{
    string name = 1 ;
}

message HelloReplay{
    string message = 1;
}

message HelloMessage{
    string msg = 1;
}

4.server代码

1.解析命令行参数
2.注册到etcd
3.监听操作系统信号（中断）

go.mod

module server

go 1.14

require (
	etcdservice v0.0.0-00010101000000-000000000000
	github.com/coreos/etcd v3.3.22+incompatible // indirect
	github.com/coreos/go-semver v0.3.0 // indirect
	github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f // indirect
	github.com/gogo/protobuf v1.3.1 // indirect
	github.com/google/uuid v1.1.1 // indirect
	go.etcd.io/etcd v3.3.22+incompatible // indirect
	go.uber.org/zap v1.15.0 // indirect
	google.golang.org/grpc v1.29.1
	protoes v0.0.0-00010101000000-000000000000
)

replace protoes => D:\projgo\grpcdemo-etcd\protoes

replace etcdservice => D:\projgo\grpcdemo-etcd\etcdservice

server.go

package main

import (
	"flag"
	"fmt"
	"etcdservice"
	pb "grpcdemo/protoes"
	"net"
	"os"
	"os/signal"
	"syscall"
	"time"

	"golang.org/x/net/context"
	"google.golang.org/grpc"
)

const (
	TIME_FORMAT = "2006-01-02 15:04:05"
)

var host = "127.0.0.1"
var (
	ServiceName = flag.String("ServiceName", "hello_service", "service name")
	Port        = flag.Int("Port", 3000, "listening port")
	EtcdAddr    = flag.String("EtcdAddr", "127.0.0.1:2379", "register etcd address")
)

// 对象要和proto内定义的服务一致
type server struct {
}

...

func main() {
	flag.Parse()
	// 监听网络
	ln, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", *Port))
	if err != nil {
		fmt.Println("网络异常：", err)
		return
	}
	defer ln.Close()

	// 创建grpc句柄
	srv := grpc.NewServer()
	defer srv.GracefulStop()

	// 将server结构体注册到grpc服务中
	pb.RegisterHelloServerServer(srv, &server{})
	addr := fmt.Sprintf("%s:%d", host, *Port)
	fmt.Printf("server addr:%s\n", addr)
  
    // 注册到etcd
	etcdservice.Register(*EtcdAddr, *ServiceName, addr, 5)

	ch := make(chan os.Signal, 1)
	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL, syscall.SIGHUP, syscall.SIGQUIT)
	go func() {
		s := <-ch
		etcdservice.UnRegister(*ServiceName, addr)

		if i, ok := s.(syscall.Signal); ok {
			os.Exit(int(i))
		} else {
			os.Exit(0)
		}

	}()

	// 监听服务，阻塞
	err = srv.Serve(ln)
	if err != nil {
		fmt.Println("监听异常：", err)
		return
	}
}

5.etcd注册服务etcdservice

1.连接etcd服务器
2.循环心跳检测

• 心跳发生器

1. KeepAlive
   • 创建租约（Lease）
   • 自动续租（KeepAlive）
   • 清空续租响应消息

3.UnRegister 并不能信赖

go.mod

module etcdservice

go 1.14

require (
	github.com/coreos/etcd v3.3.22+incompatible
	github.com/coreos/go-semver v0.3.0 // indirect
	github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f // indirect
	github.com/gogo/protobuf v1.3.1 // indirect
	github.com/google/uuid v1.1.1 // indirect
	go.uber.org/zap v1.15.0 // indirect
	google.golang.org/grpc v1.29.1
)

naming.go

package etcdservice

import (
	"context"
	"fmt"
	"log"
	"strings"
	"time"

	"github.com/coreos/etcd/clientv3"
)

// Register register service with name as prefix to etcd, multi etcd addr should use ; to split
func Register(etcdAddr, name string, addr string, ttl int64) error {
	var err error

	if cli == nil {
		cli, err = clientv3.New(clientv3.Config{
			Endpoints:   strings.Split(etcdAddr, ";"),
			DialTimeout: 15 * time.Second,
		})
		if err != nil {
			fmt.Printf("connect to etcd err:%s", err)
			return err
		}
	}

	ticker := time.NewTicker(time.Second * time.Duration(ttl))

	go func() {
		for {
			getResp, err := cli.Get(context.Background(), "/"+schema+"/"+name+"/"+addr)
			fmt.Printf("getResp:%+v\n", getResp)
			if err != nil {
				log.Println(err)
				fmt.Printf("Register:%s", err)
			} else if getResp.Count == 0 {
				err = withAlive(name, addr, ttl)
				if err != nil {
					log.Println(err)
					fmt.Printf("keep alive:%s", err)
				}
			} else {
				// fmt.Printf("getResp:%+v, do nothing\n", getResp)
			}

			<-ticker.C
		}
	}()

	return nil
}

func withAlive(name string, addr string, ttl int64) error {
	leaseResp, err := cli.Grant(context.Background(), ttl)
	if err != nil {
		return err
	}

	//fmt.Printf("key:%v\n", "/"+schema+"/"+name+"/"+addr)
	_, err = cli.Put(context.Background(), "/"+schema+"/"+name+"/"+addr, addr, clientv3.WithLease(leaseResp.ID))
	if err != nil {
		fmt.Printf("put etcd error:%s", err)
		return err
	}

	ch, err := cli.KeepAlive(context.Background(), leaseResp.ID)
	if err != nil {
		fmt.Printf("keep alive error:%s", err)
		return err
	}

	// 清空 keep alive 返回的channel
	go func() {
		for {
			<-ch
			// ka := <-ch
			// fmt.Println("ttl:", ka.TTL)
		}
	}()

	return nil
}

// UnRegister remove service from etcd
func UnRegister(name string, addr string) {
	if cli != nil {
		cli.Delete(context.Background(), "/"+schema+"/"+name+"/"+addr)
	}
}

resolver.go

package etcdservice

import (
	"context"
	"log"
	"strings"
	"time"

	"github.com/coreos/etcd/clientv3"
	// "go.etcd.io/etcd/mvcc/mvccpb"

	"github.com/coreos/etcd/mvcc/mvccpb"
	"google.golang.org/grpc/resolver"
)

const schema = "ns"

var cli *clientv3.Client

type etcdResolver struct {
	rawAddr string
	cc      resolver.ClientConn
}

// NewResolver initialize an etcd client
func NewResolver(etcdAddr string) resolver.Builder {
	return &etcdResolver{rawAddr: etcdAddr}
}

func (r *etcdResolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	var err error
	//构建etcd client
	if cli == nil {
		cli, err = clientv3.New(clientv3.Config{
			Endpoints:   strings.Split(r.rawAddr, ";"),
			DialTimeout: 15 * time.Second,
		})
		if err != nil {
			return nil, err
		}
	}

	r.cc = cc

	go r.watch("/" + target.Scheme + "/" + target.Endpoint + "/")

	return r, nil
}

func (r etcdResolver) Scheme() string {
	return schema
}

func (r etcdResolver) ResolveNow(rn resolver.ResolveNowOptions) {
	log.Println("ResolveNow") // TODO check
}

// Close closes the resolver.
func (r etcdResolver) Close() {
	log.Println("Close")
}

func (r *etcdResolver) watch(keyPrefix string) {
	var addrList []resolver.Address

	getResp, err := cli.Get(context.Background(), keyPrefix, clientv3.WithPrefix())
	if err != nil {
		log.Println(err)
	} else {
		for i := range getResp.Kvs {
			addrList = append(addrList, resolver.Address{Addr: strings.TrimPrefix(string(getResp.Kvs[i].Key), keyPrefix)})
		}
	}

	r.cc.NewAddress(addrList)

	rch := cli.Watch(context.Background(), keyPrefix, clientv3.WithPrefix())
	for n := range rch {
		for _, ev := range n.Events {
			addr := strings.TrimPrefix(string(ev.Kv.Key), keyPrefix)
			switch ev.Type {
			case mvccpb.PUT:
				if !exist(addrList, addr) {
					addrList = append(addrList, resolver.Address{Addr: addr})
					r.cc.NewAddress(addrList)
				}
			case mvccpb.DELETE:
				if s, ok := remove(addrList, addr); ok {
					addrList = s
					r.cc.NewAddress(addrList)
				}
			}
			//log.Printf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
		}
	}
}

func exist(l []resolver.Address, addr string) bool {
	for i := range l {
		if l[i].Addr == addr {
			return true
		}
	}
	return false
}

func remove(s []resolver.Address, addr string) ([]resolver.Address, bool) {
	for i := range s {
		if s[i].Addr == addr {
			s[i] = s[len(s)-1]
			return s[:len(s)-1], true
		}
	}
	return nil, false
}

6.client代码

client实现

创建etcd解析器（Resolver）
解析器注册到grpc/resolver
使用名称拨号: scheme://authority/endpoint_name
应用负载均衡策略
获取服务端连接
改造代码，无限循环调用服务

go.mod

module client

go 1.14

replace protoes => D:\projgo\grpcdemo-etcd\protoes

replace etcdservice => D:\projgo\grpcdemo-etcd\etcdservice

require (
	etcdservice v0.0.0-00010101000000-000000000000
	google.golang.org/grpc v1.29.1
	protoes v0.0.0-00010101000000-000000000000
)

client.go

package main

import (
	"context"
	"etcdservice"
	"flag"
	"fmt"
	pb "protoes"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/resolver"
)

var (
	ServiceName = flag.String("ServiceName", "hello_service", "service name")
	EtcdAddr    = flag.String("EtcdAddr", "127.0.0.1:2379", "register etcd address")
)

func main() {
	flag.Parse()
	r := etcdservice.NewResolver(*EtcdAddr)
	resolver.Register(r)

	// 客户端连接服务器
	// conn, err := grpc.Dial("127.0.0.1:3000", grpc.WithInsecure())
	conn, err := grpc.Dial(r.Scheme()+"://author/"+*ServiceName, grpc.WithBalancerName("round_robin"), grpc.WithInsecure())

	if err != nil {
		fmt.Println("连接服务器失败", err)
	}
	defer conn.Close()

	// 获得grpc句柄
	c := pb.NewHelloServerClient(conn)
	ticker := time.NewTicker(1 * time.Second)
	for t := range ticker.C {
		// 远程单调用 SayHi 接口
		r1, err := c.SayHi(
			context.Background(),
			&pb.HelloRequest{
				Name: "Kitty",
			},
		)
		if err != nil {
			fmt.Println("Can not get SayHi:", err)
			return
		}
		fmt.Printf("%v: SayHi 响应：%s\n", t, r1.GetMessage())

		// 远程单调用 GetMsg 接口
		r2, err := c.GetMsg(
			context.Background(),
			&pb.HelloRequest{
				Name: "Kitty",
			},
		)
		if err != nil {
			fmt.Println("Can not get GetMsg:", err)
			return
		}
		fmt.Printf("%v: GetMsg 响应：%s\n", t, r2.GetMsg())
	}

}

7.etcd名称解析器

1.类成员设计，全局变量
2.实现Build方法

• 连接etcd服务器
• 保存连接给watch使用
• 启动watch

package main

import (
	"context"
	"fmt"
	"time"

	"go.etcd.io/etcd/clientv3"
)

func main() {
	var (
		config        clientv3.Config
		client        *clientv3.Client
		lease         clientv3.Lease
		leaseResp     *clientv3.LeaseGrantResponse
		leaseId       clientv3.LeaseID
		leaseRespChan <-chan *clientv3.LeaseKeepAliveResponse
		err           error
	)
	//客户端配置
	config = clientv3.Config{
		Endpoints:   []string{"127.0.0.1:2379"},
		DialTimeout: 5 * time.Second,
	}
	//建立连接
	if client, err = clientv3.New(config); err != nil {
		fmt.Println(err)
		return
	}

	//上锁（创建租约，自动续租）
	lease = clientv3.NewLease(client)

	//设置一个ctx取消自动续租
	ctx, cancleFunc := context.WithCancel(context.TODO())

	//设置10秒租约（过期时间）
	if leaseResp, err = lease.Grant(context.TODO(), 10); err != nil {
		fmt.Println(err)
		return
	}
	//拿到租约id
	leaseId = leaseResp.ID

	//自动续租（不停地往管道中扔租约信息）
	if leaseRespChan, err = lease.KeepAlive(ctx, leaseId); err != nil {
		fmt.Println(err)

	}
	//启动一个协程去监听
	go listenLeaseChan(leaseRespChan)
	//业务处理
	kv := clientv3.NewKV(client)
	//创建事务
	txn := kv.Txn(context.TODO())
	txn.If(clientv3.Compare(clientv3.CreateRevision("/lock/resource1"), "=", 0)).
		Then(clientv3.OpPut("/lock/resource1", "true", clientv3.WithLease(leaseId))).
		Else(clientv3.OpGet("/lock/resource1")) //否则抢锁失败

	//提交事务
	if txtResp, err := txn.Commit(); err != nil {
		fmt.Println(err)
		return
	} else {
		//判断是否抢锁
		if !txtResp.Succeeded {
			fmt.Println("锁被占用：", string(txtResp.Responses[0].GetResponseRange().Kvs[0].Value))
			return
		}
	}

	fmt.Println("处理任务")

	//释放锁（停止续租，终止租约）
	defer cancleFunc()                          //函数退出取消自动续租
	defer lease.Revoke(context.TODO(), leaseId) //终止租约（去掉过期时间）

	time.Sleep(10 * time.Second)

}

func listenLeaseChan(leaseRespChan <-chan *clientv3.LeaseKeepAliveResponse) {
	var leaseKeepResp *clientv3.LeaseKeepAliveResponse
	for {
		select {
		case leaseKeepResp = <-leaseRespChan:
			if leaseKeepResp == nil {
				fmt.Println("租约失效了")
				goto END
			} else {
				fmt.Println(leaseKeepResp.ID)

			}

		}
	}
END:
}

8.Watch 说明

1.初始化

• 监控指定目录
• 获取服务器列表
• 添加地址列表到gRPC连接
  2.监控变化
• 服务记录添加
• 服务记录删除

9.运行演示

在三个终端窗口启动服务

server.exe -Port 3000
server.exe -Port 3001
server.exe -Port 3002

ETCD Manager 客户端观察etcd状态

启动客户端

观察服务器窗口

1. 三个服务轮流收到消息，负载均衡
2. 停止其中的服务，并重启，自动重新均衡访问
3. 停止所有服务，客户端退出

etcd重启，不中断客户端访问服务端

10.总结

etcd-golang

• 服务端注册
• 注册两重循环，心跳，租约
• 客户端拨号使用的名称协议
• 名称解析实现，build，watch