golang grpc 负载均衡的方法

微服务架构里面，每个服务都会有很多节点，如果流量分配不均匀，会造成资源的浪费，甚至将一些机器压垮，这个时候就需要负载均衡，最简单的一种策略就是轮询，顺序依次选择不同的节点访问。

grpc 在客户端提供了负载均衡的实现，并提供了服务地址解析和更新的接口(默认提供了 dns 域名解析的支持)，方便不同服务的集成

使用示例

conn, err := grpc.dial(
  "",
  grpc.withinsecure(),
  // 负载均衡，使用 consul 作服务发现
  grpc.withbalancer(grpc.roundrobin(grpclb.newconsulresolver(
    "127.0.0.1:8500", "grpc.health.v1.add",
  ))),
)

创建连接的时候可以使用 withbalancer 选项来指定负载均衡策略，这里使用 roundrobin 算法，其实就是轮询策略

与 consul 的集成

有了负载均衡策略，还需要一个地址解析和更新策略，可以使用 dns 服务来实现，但如果我们使用 consul 来做服务的注册和发现，可以通过实现 ‘naming.resolver' 和 ‘naming.watcher' 接口来支持

• naming.resolver: 实现地址解析
• naming.watcher: 实现节点的变更，添加或者删除

func newconsulresolver(address string, service string) naming.resolver {
  return &consulresolver{
    address: address,
    service: service,
  }
}
type consulresolver struct {
  address string
  service string
}
func (r *consulresolver) resolve(target string) (naming.watcher, error) {
  config := api.defaultconfig()
  config.address = r.address
  client, err := api.newclient(config)
  if err != nil {
    return nil, err
  }
  return &consulwatcher{
    client: client,
    service: r.service,
    addrs:  map[string]struct{}{},
  }, nil
}
type consulwatcher struct {
  client  *api.client
  service  string
  addrs   map[string]struct{}
  lastindex uint64
}
func (w *consulwatcher) next() ([]*naming.update, error) {
  for {
    services, metainfo, err := w.client.health().service(w.service, "", true, &api.queryoptions{
      waitindex: w.lastindex, // 同步点，这个调用将一直阻塞，直到有新的更新
    })
    if err != nil {
      logrus.warn("error retrieving instances from consul: %v", err)
    }
    w.lastindex = metainfo.lastindex
    addrs := map[string]struct{}{}
    for _, service := range services {
      addrs[net.joinhostport(service.service.address, strconv.itoa(service.service.port))] = struct{}{}
    }
    var updates []*naming.update
    for addr := range w.addrs {
      if _, ok := addrs[addr]; !ok {
        updates = append(updates, &naming.update{op: naming.delete, addr: addr})
      }
    }
    for addr := range addrs {
      if _, ok := w.addrs[addr]; !ok {
        updates = append(updates, &naming.update{op: naming.add, addr: addr})
      }
    }
    if len(updates) != 0 {
      w.addrs = addrs
      return updates, nil
    }
  }
}
func (w *consulwatcher) close() {
  // nothing to do
}

参考链接

grpc name resolution:

load balancing in grpc:

dns_resolver:

代码地址:

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。