开发者社区> OpenIM> 正文

OpenIM服务发现和负载均衡golang插件:gRPC接入etcdv3

简介: OpenIM集成此插件,实现了轻量级的服务发现机制,打造了基于集群的IM服务,各模块很方便平行扩展,方便运维,本文把思路和代码分享给大家。
+关注继续查看

etcd简介

etcd是CoreOS团队于2013年6月发起的开源项目,它的目标是构建一个高可用的分布式键值(key-value)数据库。etcd内部采用raft协议作为一致性算法,etcd基于Go语言实现。

etcd作为服务发现系统,有以下的特点:

  • 简单:安装配置简单,而且提供了HTTP API进行交互,使用也很简单
  • 安全:支持SSL证书验证
  • 快速:根据官方提供的benchmark数据,单实例支持每秒2k+读操作
  • 可靠:采用raft算法,实现分布式系统数据的可用性和一致性

etcd项目地址:https://github.com/coreos/etcd/

etcd典型应用场景-服务发现

etcd比较多的应用场景是用于服务发现,服务发现(Service Discovery)要解决的是分布式系统中最常见的问题之一,即在同一个分布式集群中的进程或服务如何才能找到对方并建立连接。

从本质上说,服务发现就是要了解集群中是否有进程在监听upd或者tcp端口,并且通过名字就可以进行查找和链接。

要解决服务发现的问题,需要下面三大支柱,缺一不可。

  • 一个强一致性、高可用的服务存储目录。

基于Ralf算法的etcd天生就是这样一个强一致性、高可用的服务存储目录。

  • 一种注册服务和健康服务健康状况的机制。

用户可以在etcd中注册服务,并且对注册的服务配置key TTL,定时保持服务的心跳以达到监控健康状态的效果。

  • 一种查找和连接服务的机制。

通过在etcd指定的主题下注册的服务业能在对应的主题下查找到。为了确保连接,我们可以在每个服务机器上都部署一个proxy模式的etcd,这样就可以确保访问etcd集群的服务都能够互相连接。

1.png

gRPC简介

gRPC是谷歌开源的一款跨平台、高性能的RPC框架。gRPC是一个现代的开源高性能RPC框架,可以在任何环境下运行。在实际开发过程中,主要使用它来进行后端微服务的开发。

在gRPC中,客户端应用程序可以像本地对象那样直接调用另一台计算机上的服务器应用程序上的方法,从而更容易创建分布式应用程序和服务。与许多RPC系统一样,gRPC基于定义服务的思想,可以通过设置参数和返回类型来远程调用方法。在服务端,实现这个接口并运行gRPC服务器来处理客户端调用。客户端提供的方法(客户端与服务端的方法相同)。

如图所示,gRPC客户端和服务端可以在各种环境中运行和相互通信,并且可以用gRPC支持的任何语言编写。因此,可以用Go语言创建一个gRPC服务器,同时供PHP客户端和Android客户端等多个客户端调用,从而突破开发语言的限制。

2.jpg

服务注册:register.go

``

/etcdAddr separated by commas

func RegisterEtcd(schema, etcdAddr, myHost string, myPort int, serviceName string, ttl int) error {

  cli, err := clientv3.New(clientv3.Config{

     Endpoints: strings.Split(etcdAddr, ","),

  })

  fmt.Println("RegisterEtcd")

  if err != nil {

     //    return fmt.Errorf("grpclb: create clientv3 client failed: %v", err)

     return fmt.Errorf("create etcd clientv3 client failed, errmsg:%v, etcd addr:%s", err, etcdAddr)

  }

  //lease

  ctx, cancel := context.WithCancel(context.Background())

  resp, err := cli.Grant(ctx, int64(ttl))

  if err != nil {

     return fmt.Errorf("grant failed")

  }

  //  schema:///serviceName/ip:port ->ip:port

  serviceValue := net.JoinHostPort(myHost, strconv.Itoa(myPort))

  serviceKey := GetPrefix(schema, serviceName) + serviceValue

  //set key->value

  if _, err := cli.Put(ctx, serviceKey, serviceValue, clientv3.WithLease(resp.ID)); err != nil {

     return fmt.Errorf("put failed, errmsg:%v, key:%s, value:%s", err, serviceKey, serviceValue)

  }

  //keepalive

  kresp, err := cli.KeepAlive(ctx, resp.ID)

  if err != nil {

     return fmt.Errorf("keepalive faild, errmsg:%v, lease id:%d", err, resp.ID)

  }

  go func() {

  FLOOP:

     for {

        select {

        case _, ok := <-kresp:

           if ok == true {

           } else {

              break FLOOP

           }

        }

     }

  }()

  rEtcd = &RegEtcd{ctx: ctx,

     cli:    cli,

     cancel: cancel,

     key:    serviceKey}

  return nil

}

grpc模块在启动时调用RegisterEtcd注册,并定时lease

命名解析实现及服务发现:resolver.go

``

type Resolver struct {

  cc                 resolver.ClientConn

  serviceName        string

  grpcClientConn     *grpc.ClientConn

  cli                *clientv3.Client

  schema             string

  etcdAddr           string

  watchStartRevision int64

}

var (

  nameResolver        = make(map[string]*Resolver)

  rwNameResolverMutex sync.RWMutex

)

func NewResolver(schema, etcdAddr, serviceName string) (*Resolver, error) {

  etcdCli, err := clientv3.New(clientv3.Config{

     Endpoints: strings.Split(etcdAddr, ","),

  })

  if err != nil {

     return nil, err

  }

  var r Resolver

  r.serviceName = serviceName

  r.cli = etcdCli

  r.schema = schema

  r.etcdAddr = etcdAddr

  resolver.Register(&r)

  conn, err := grpc.Dial(

     GetPrefix(schema, serviceName),

     grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, roundrobin.Name)),

     grpc.WithInsecure(),

     grpc.WithTimeout(time.Duration(5)*time.Second),

  )

  if err == nil {

     r.grpcClientConn = conn

  }

  return &r, err

}

func (r1 *Resolver) ResolveNow(rn resolver.ResolveNowOptions) {

}

func (r1 *Resolver) Close() {

}

func GetConn(schema, etcdaddr, serviceName string) *grpc.ClientConn {

  rwNameResolverMutex.RLock()

  r, ok := nameResolver[schema+serviceName]

  rwNameResolverMutex.RUnlock()

  if ok {

     return r.grpcClientConn

  }

  rwNameResolverMutex.Lock()

  r, ok = nameResolver[schema+serviceName]

  if ok {

     rwNameResolverMutex.Unlock()

     return r.grpcClientConn

  }

  r, err := NewResolver(schema, etcdaddr, serviceName)

  if err != nil {

     rwNameResolverMutex.Unlock()

     return nil

  }

  nameResolver[schema+serviceName] = r

  rwNameResolverMutex.Unlock()

  return r.grpcClientConn

}

func (r *Resolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {

  if r.cli == nil {

     return nil, fmt.Errorf("etcd clientv3 client failed, etcd:%s", target)

  }

  r.cc = cc

  ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)

  //     "%s:///%s"

  prefix := GetPrefix(r.schema, r.serviceName)

  // get key first

  resp, err := r.cli.Get(ctx, prefix, clientv3.WithPrefix())

  if err == nil {

     var addrList []resolver.Address

     for i := range resp.Kvs {

        fmt.Println("init addr: ", string(resp.Kvs[i].Value))

        addrList = append(addrList, resolver.Address{Addr: string(resp.Kvs[i].Value)})

     }

     r.cc.UpdateState(resolver.State{Addresses: addrList})

     r.watchStartRevision = resp.Header.Revision + 1

     go r.watch(prefix, addrList)

  } else {

     return nil, fmt.Errorf("etcd get failed, prefix: %s", prefix)

  }

  return r, nil

}

func (r *Resolver) Scheme() string {

  return r.schema

}

func exists(addrList []resolver.Address, addr string) bool {

  for _, v := range addrList {

     if v.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

}

func (r *Resolver) watch(prefix string, addrList []resolver.Address) {

  rch := r.cli.Watch(context.Background(), prefix, clientv3.WithPrefix(), clientv3.WithPrefix())

  for n := range rch {

     flag := 0

     for _, ev := range n.Events {

        switch ev.Type {

        case mvccpb.PUT:

           if !exists(addrList, string(ev.Kv.Value)) {

              flag = 1

              addrList = append(addrList, resolver.Address{Addr: string(ev.Kv.Value)})

              fmt.Println("after add, new list: ", addrList)

           }

        case mvccpb.DELETE:

           fmt.Println("remove addr key: ", string(ev.Kv.Key), "value:", string(ev.Kv.Value))

           i := strings.LastIndexAny(string(ev.Kv.Key), "/")

           if i < 0 {

              return

           }

           t := string(ev.Kv.Key)[i+1:]

           fmt.Println("remove addr key: ", string(ev.Kv.Key), "value:", string(ev.Kv.Value), "addr:", t)

           if s, ok := remove(addrList, t); ok {

              flag = 1

              addrList = s

              fmt.Println("after remove, new list: ", addrList)

           }

        }

     }

     if flag == 1 {

        r.cc.UpdateState(resolver.State{Addresses: addrList})

        fmt.Println("update: ", addrList)

     }

  }

}

客户端先通过GetConn获取conn,然后再调用grpc服务,调用后不用关闭conn

服务端示例代码:server.go

``

getcdv3.RegisterEtcd ("sk", etcdAddr, "127.0.0.1", port, "myrpc1", 10)

s := grpc.NewServer()

helloworld.RegisterHelloServer(s, &server{})

s.Serve(listener)

客户端示例代码:client.go

``

p := getcdv3.GetConn("sk", etcdAddr, "myrpc1")

client := helloworld.NewHelloClient(p)

resp1, err := client.SayHello(context.Background(), &helloworld.HelloReq{Req: "world"})

总结:OpenIM集成此插件,实现了轻量级的服务发现机制,打造了基于集群的IM服务,各模块很方便平行扩展,方便运维。在使用grpc、etcd过程中,特别注意版本兼容问题,具体可以参考OpenIM的go.mod文件


更多阅读

基于Tablestore Timeline的IM(即时通讯)消息系统架构 - 架构篇

OpenIM官网

【原创】开源OpenIM:高性能、可伸缩、易扩展的即时通讯架构

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
使用OpenApi弹性管理云服务器ECS
阿里云的云服务器ECS除了提供控制台来进行日常的管理和资源创建,还提供了OpenApi来进行资源的管理和定制开发。通过OpenApi您可以更加灵活的管理和配置云服务器。 阿里云提供了SDK来包装OpenApi,可以让您将云服务器的管理集成到您的已有系统中。
10642 0
关于wordpress主题、插件上传和下载问题及其上传图片权限问题解决方案
主题官方下载地址:https://wordpress.org/themes/ 插件官方下载地址: https://wordpress.org/plugins/ 主题的上传下载,无疑是需要ftp服务器的。
2025 0
docker服务发现——etcd集群
etcd etcd用于服务发现的基础注册和通知,功能类似于zk,通过注册和监听,实现基础的服务发现。 安装 etcd安装非常简单,可以用go自己编译,etcd也提供了可以直接使用的二进制包(64位)。 具体的安装提示页面在github上, 直接按照上面的描述下载即可。为了方便,把里面的et
10791 0
vscode插件快餐教程(4) - 语言服务器协议lsp
在有lsp之前,存在三个主要问题: 一是语言相关的扩展都是用该语言母语写的,不容易集成到插件中去。毕竟现在大量的语言都带有运行时。 二是语言扫描相关的工作都比较占用CPU资源,运行在vscode内部不如放在独立进程,甚至远程服务器上更好。
2991 0
使用 grpcurl 通过命令行访问 gRPC 服务
使用 grpcurl 通过命令行访问 gRPC 服务
33 0
Openfire服务器配置(Windows)
源码配置完成后,输入http://127.0.0.1:9090/即可打开Openfire服务器后台管理页。 设置语言。点击Continue,建议配置时切勿选择中文,否则后面的配置中可能会出现乱码。
1121 0
+关注
OpenIM
开源IM框架OpenIM的作者 前微信高级架构师
7
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
OceanBase 入门到实战教程
立即下载
阿里云图数据库GDB,加速开启“图智”未来.ppt
立即下载
实时数仓Hologres技术实战一本通2.0版(下)
立即下载