热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

ETCD组件在grpc中的实践

一、前言grpc中没有像go-micro那样集成可插拔式的etcd库使用,如何使得grpc能够使用服务注册发现及命名解析的功能,因此本文基于etcd实现了NameResolver。

一、前言

grpc中没有像go-micro那样集成可插拔式的etcd库使用,如何使得grpc能够使用服务注册发现及命名解析的功能,因此本文基于etcd实现了Name Resolver。

 

二、所需的grpc版本及高版本grpc、protobuf与etcd兼容问题

grpc相关库:

google.golang.org/grpc v1.26.0

google.golang.org/grpc/balancer/roundrobin

google.golang.org/grpc/resolver

 

etcd相关库:

go.etcd.io/etcd/clientv3

github.com/coreos/etcd/mvcc/mvccpb

 

此处需要注意的是,新版本grpc不兼容etcd相关库, 如果grpc版本大于1.26.0或者protobuf版本过高会出现以下问题:

1. grpc版本过高,新版本不支持etcd  需降级

# github.com/coreos/etcd/clientv3/balancer/picker
../../vendor/github.com/coreos/etcd/clientv3/balancer/picker/err.go:37:44: undefined: balancer.PickOptions
../../vendor/github.com/coreos/etcd/clientv3/balancer/picker/roundrobin_balanced.go:55:54: undefined: balancer.PickOptions
# github.com/coreos/etcd/clientv3/balancer/resolver/endpoint
../../vendor/github.com/coreos/etcd/clientv3/balancer/resolver/endpoint/endpoint.go:114:78: undefined: resolver.BuildOption
../../vendor/github.com/coreos/etcd/clientv3/balancer/resolver/endpoint/endpoint.go:182:31: undefined: resolver.ResolveNowOption

解决办法:在go.mod 加入:replace google.golang.org/grpc => google.golang.org/grpc v1.26.0

2. grpc版本过低  protobuf版本过高

# ut-blogger/api/protos/user
../../api/protos/user/user.pb.go:327:11: undefined: grpc.SupportPackageIsVersion6
../../api/protos/user/user.pb.go:338:5: undefined: grpc.ClientConnInterface

解决办法:降级protoc-gen-go的版本 go get [email protected] 重新生成proto

 

三、自定义实现naming

package etcdservice
import (
"context"
"go.etcd.io/etcd/clientv3"
"log"
"strings"
"time"
)
const Schema = "grpcEtcd"
// Register 注册地址到ETCD组件中 使用 ; 分割
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 {
log.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)
if err != nil {
log.Printf("getResp:%+v\n", getResp)
log.Printf("Register:%s", err)
} else if getResp.Count == 0 {
err = withAlive(name, addr, ttl)
if err != nil {
log.Printf("keep alive:%s", err)
}
}
<-ticker.C
}
}()
return nil
}
// withAlive 创建租约
func withAlive(name string, addr string, ttl int64) error {
leaseResp, err := cli.Grant(context.Background(), ttl)
if err != nil {
return err
}
log.Printf("key:%v\n", "/"+Schema+"/"+name+"/"+addr)
_, err = cli.Put(context.Background(), "/"+Schema+"/"+name+"/"+addr, addr, clientv3.WithLease(leaseResp.ID))
if err != nil {
log.Printf("put etcd error:%s", err)
return err
}
ch, err := cli.KeepAlive(context.Background(), leaseResp.ID)
if err != nil {
log.Printf("keep alive error:%s", err)
return err
}
// 清空 keep alive 返回的channel
go func() {
for {
<-ch
}
}()
return nil
}
// UnRegister remove service from etcd
func UnRegister(name string, addr string) {
if cli != nil {
cli.Delete(context.Background(), "/"+Schema+"/"+name+"/"+addr)
}
}

 

四、自定义实现resolver

package etcdservice
import (
"context"
"log"
"strings"
"time"
"go.etcd.io/etcd/clientv3"
"github.com/coreos/etcd/mvcc/mvccpb"
"google.golang.org/grpc/resolver"
)
var cli *clientv3.Client
// etcdResolver 解析struct
type etcdResolver struct {
rawAddr string
cc resolver.ClientConn
}
// NewResolver initialize an etcd client
func NewResolver(etcdAddr string) resolver.Builder {
return &etcdResolver{rawAddr: etcdAddr}
}
// Build 构建etcd client
func (r *etcdResolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
var err error
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
}
// Scheme etcd resolve scheme
func (r etcdResolver) Scheme() string {
return Schema
}
// ResolveNow
func (r etcdResolver) ResolveNow(rn resolver.ResolveNowOptions) {
log.Println("ResolveNow")
}
// Close closes the resolver
func (r etcdResolver) Close() {
log.Println("Close")
}
// watch 监听resolve列表变化
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)})
}
}
// 新版本etcd去除了NewAddress方法 以UpdateState代替
r.cc.UpdateState(resolver.State{Addresses: 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.UpdateState(resolver.State{Addresses: addrList})
}
case mvccpb.DELETE:
if s, ok := remove(addrList, addr); ok {
addrList = s
r.cc.UpdateState(resolver.State{Addresses: addrList})
}
}
log.Printf("%s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
}
}
}
// exist 判断resolve address是否存在
func exist(l []resolver.Address, addr string) bool {
for i := range l {
if l[i].Addr == addr {
return true
}
}
return false
}
// remove 从resolver列表移除
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
}

 

五、服务端调用ETCD服务注册

EtcdAddr、ServiceName和Ttl可以从配置文件读取,读取配置文件的方式很多,本文不在此阐述

//将服务地址注册到etcd中
go etcdservice.Register(s.o.EtcdAddr, s.o.ServiceName, addr, s.o.Ttl)
ch := make(chan os.Signal, 1)
signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL, syscall.SIGHUP, syscall.SIGQUIT)
go func() {
sig := <-ch
etcdservice.UnRegister(s.o.ServiceName, addr)
if i, ok := sig.(syscall.Signal); ok {
os.Exit(int(i))
} else {
os.Exit(0)
}
}()

 

六、 客户端grpc服务 ectd服务发现

客户端调用grpc服务以serviceName的形式从etcd中获取服务节点,此处采用roundrobin的形式作为负载均衡

//注册etcd解析器
r := etcdservice.NewResolver(o.EtcdAddr)
resolver.Register(r)
// 客户端连接服务器
conn, err := grpc2.Dial(r.Scheme()+"://"+o.Caller+"/"+o.Callee, grpc2.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, roundrobin.Name)), grpc2.WithInsecure())
if err != nil {
log.Println("连接服务器失败", err)
return nil, errors.Wrap(err, "notify client dial error")
}

 



推荐阅读
  • 原文网址:https:www.cnblogs.comysoceanp7476379.html目录1、AOP什么?2、需求3、解决办法1:使用静态代理4 ... [详细]
  • Spring Boot 中配置全局文件上传路径并实现文件上传功能
    本文介绍如何在 Spring Boot 项目中配置全局文件上传路径,并通过读取配置项实现文件上传功能。通过这种方式,可以更好地管理和维护文件路径。 ... [详细]
  • Hadoop的文件操作位于包org.apache.hadoop.fs里面,能够进行新建、删除、修改等操作。比较重要的几个类:(1)Configurati ... [详细]
  • 本文详细介绍了Java反射机制的基本概念、获取Class对象的方法、反射的主要功能及其在实际开发中的应用。通过具体示例,帮助读者更好地理解和使用Java反射。 ... [详细]
  • javax.mail.search.BodyTerm.matchPart()方法的使用及代码示例 ... [详细]
  • Spring – Bean Life Cycle
    Spring – Bean Life Cycle ... [详细]
  • DAO(Data Access Object)模式是一种用于抽象和封装所有对数据库或其他持久化机制访问的方法,它通过提供一个统一的接口来隐藏底层数据访问的复杂性。 ... [详细]
  • 本文介绍了在 Java 编程中遇到的一个常见错误:对象无法转换为 long 类型,并提供了详细的解决方案。 ... [详细]
  • 在多线程并发环境中,普通变量的操作往往是线程不安全的。本文通过一个简单的例子,展示了如何使用 AtomicInteger 类及其核心的 CAS 无锁算法来保证线程安全。 ... [详细]
  • 实验九:使用SharedPreferences存储简单数据
    本实验旨在帮助学生理解和掌握使用SharedPreferences存储和读取简单数据的方法,包括程序参数和用户选项。 ... [详细]
  • 字节流(InputStream和OutputStream),字节流读写文件,字节流的缓冲区,字节缓冲流
    字节流抽象类InputStream和OutputStream是字节流的顶级父类所有的字节输入流都继承自InputStream,所有的输出流都继承子OutputStreamInput ... [详细]
  • 本文将详细介绍如何注册码云账号、配置SSH公钥、安装必要的开发工具,并逐步讲解如何下载、编译 HarmonyOS 2.0 源码。通过本文,您将能够顺利完成 HarmonyOS 2.0 的环境搭建和源码编译。 ... [详细]
  • 本文介绍了如何使用 Node.js 和 Express(4.x 及以上版本)构建高效的文件上传功能。通过引入 `multer` 中间件,可以轻松实现文件上传。首先,需要通过 `npm install multer` 安装该中间件。接着,在 Express 应用中配置 `multer`,以处理多部分表单数据。本文详细讲解了 `multer` 的基本用法和高级配置,帮助开发者快速搭建稳定可靠的文件上传服务。 ... [详细]
  • 本文详细解析了客户端与服务器之间的交互过程,重点介绍了Socket通信机制。IP地址由32位的4个8位二进制数组成,分为网络地址和主机地址两部分。通过使用 `ipconfig /all` 命令,用户可以查看详细的IP配置信息。此外,文章还介绍了如何使用 `ping` 命令测试网络连通性,例如 `ping 127.0.0.1` 可以检测本机网络是否正常。这些技术细节对于理解网络通信的基本原理具有重要意义。 ... [详细]
  • ### 优化后的摘要本学习指南旨在帮助读者全面掌握 Bootstrap 前端框架的核心知识点与实战技巧。内容涵盖基础入门、核心功能和高级应用。第一章通过一个简单的“Hello World”示例,介绍 Bootstrap 的基本用法和快速上手方法。第二章深入探讨 Bootstrap 与 JSP 集成的细节,揭示两者结合的优势和应用场景。第三章则进一步讲解 Bootstrap 的高级特性,如响应式设计和组件定制,为开发者提供全方位的技术支持。 ... [详细]
author-avatar
就是!有梦想
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有