balancer.go
2024-10-13 18:30:08
package clientv3
import (
"net/url"
"strings"
"sync"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
)
// ErrNoAddrAvilable is returned by Get() when the balancer does not have
// any active connection to endpoints at the time.
// This error is returned only when opts.BlockingWait is true.
var ErrNoAddrAvilable = grpc.Errorf(codes.Unavailable, "there is no address available")
// simpleBalancer does the bare minimum to expose multiple eps
// to the grpc reconnection code path
//简单均衡器
type simpleBalancer struct {
// addrs are the client's endpoints for grpc
addrs []grpc.Address //nsqd 地址列表
// notifyCh notifies grpc of the set of addresses for connecting
notifyCh chan []grpc.Address //链接通知地址
// readyc closes once the first connection is up
//一旦连接上 ,就关闭链接 并且 只执行一次
readyc chan struct{}
readyOnce sync.Once
// mu protects upEps, pinAddr, and connectingAddr
//锁 保护 upEps, pinAddr, and connectingAddr 的锁
mu sync.RWMutex
// upEps holds the current endpoints that have an active connection
//已经存活的链接地址
upEps map[string]struct{}
// upc closes when upEps transitions from empty to non-zero or the balancer closes.
//当upEps为空 转化为非空 或者均衡器关闭的时候 ,关闭此通道
upc chan struct{}
// grpc issues TLS cert checks using the string passed into dial so
// that string must be the host. To recover the full scheme://host URL,
// have a map from hosts to the original endpoint.
//host 到 endpoint 映射 map集合
host2ep map[string]string
// pinAddr is the currently pinned address; set to the empty string on
// intialization and shutdown.
//当前固定的地址。当此变量被初始化或者关闭的时候 设置为空
pinAddr string
//是否关闭的标志
closed bool
}
//创建负载均衡器
func newSimpleBalancer(eps []string) *simpleBalancer {
notifyCh := make(chan []grpc.Address, 1)
addrs := make([]grpc.Address, len(eps))
for i := range eps {
addrs[i].Addr = getHost(eps[i])
}
notifyCh <- addrs
sb := &simpleBalancer{
addrs: addrs,
notifyCh: notifyCh,
readyc: make(chan struct{}),
upEps: make(map[string]struct{}),
upc: make(chan struct{}),
host2ep: getHost2ep(eps),
}
return sb
}
//启动
func (b *simpleBalancer) Start(target string, config grpc.BalancerConfig) error { return nil }
//链接通知
func (b *simpleBalancer) ConnectNotify() <-chan struct{} {
b.mu.Lock()
defer b.mu.Unlock()
return b.upc
}
//通过主机 转化为地址
func (b *simpleBalancer) getEndpoint(host string) string {
b.mu.Lock()
defer b.mu.Unlock()
return b.host2ep[host]
}
//同上相反
func getHost2ep(eps []string) map[string]string {
hm := make(map[string]string, len(eps))
for i := range eps {
_, host, _ := parseEndpoint(eps[i])
hm[host] = eps[i]
}
return hm
}
//更新地址列表
func (b *simpleBalancer) updateAddrs(eps []string) {
np := getHost2ep(eps)
b.mu.Lock()
defer b.mu.Unlock()
match := len(np) == len(b.host2ep)
for k, v := range np {
if b.host2ep[k] != v {
match = false
break
}
}
if match {
// same endpoints, so no need to update address
return
}
b.host2ep = np
addrs := make([]grpc.Address, 0, len(eps))
for i := range eps {
addrs = append(addrs, grpc.Address{Addr: getHost(eps[i])})
}
b.addrs = addrs
b.notifyCh <- addrs
}
//运行中的机器
func (b *simpleBalancer) Up(addr grpc.Address) func(error) {
b.mu.Lock()
defer b.mu.Unlock()
// gRPC might call Up after it called Close. We add this check
// to "fix" it up at application layer. Or our simplerBalancer
// might panic since b.upc is closed.
if b.closed {
return func(err error) {}
}
if len(b.upEps) == 0 {
// notify waiting Get()s and pin first connected address
close(b.upc)
b.pinAddr = addr.Addr
}
b.upEps[addr.Addr] = struct{}{}
// notify client that a connection is up
b.readyOnce.Do(func() { close(b.readyc) })
return func(err error) {
b.mu.Lock()
delete(b.upEps, addr.Addr)
if len(b.upEps) == 0 && b.pinAddr != "" {
b.upc = make(chan struct{})
} else if b.pinAddr == addr.Addr {
// choose new random up endpoint
for k := range b.upEps {
b.pinAddr = k
break
}
}
b.mu.Unlock()
}
}
func (b *simpleBalancer) Get(ctx context.Context, opts grpc.BalancerGetOptions) (grpc.Address, func(), error) {
var addr string
// If opts.BlockingWait is false (for fail-fast RPCs), it should return
// an address it has notified via Notify immediately instead of blocking.
if !opts.BlockingWait {
b.mu.RLock()
closed := b.closed
addr = b.pinAddr
upEps := len(b.upEps)
b.mu.RUnlock()
if closed {
return grpc.Address{Addr: ""}, nil, grpc.ErrClientConnClosing
}
if upEps == 0 {
return grpc.Address{Addr: ""}, nil, ErrNoAddrAvilable
}
return grpc.Address{Addr: addr}, func() {}, nil
}
for {
b.mu.RLock()
ch := b.upc
b.mu.RUnlock()
select {
case <-ch:
case <-ctx.Done():
return grpc.Address{Addr: ""}, nil, ctx.Err()
}
b.mu.RLock()
addr = b.pinAddr
upEps := len(b.upEps)
b.mu.RUnlock()
if addr == "" {
return grpc.Address{Addr: ""}, nil, grpc.ErrClientConnClosing
}
if upEps > 0 {
break
}
}
return grpc.Address{Addr: addr}, func() {}, nil
}
func (b *simpleBalancer) Notify() <-chan []grpc.Address { return b.notifyCh }
func (b *simpleBalancer) Close() error {
b.mu.Lock()
defer b.mu.Unlock()
// In case gRPC calls close twice. TODO: remove the checking
// when we are sure that gRPC wont call close twice.
if b.closed {
return nil
}
b.closed = true
close(b.notifyCh)
// terminate all waiting Get()s
b.pinAddr = ""
if len(b.upEps) == 0 {
close(b.upc)
}
return nil
}
func getHost(ep string) string {
url, uerr := url.Parse(ep)
if uerr != nil || !strings.Contains(ep, "://") {
return ep
}
return url.Host
}
最新文章
- Hibernate基本CRUD
- xargs
- HTML页面导航栏页脚不动,变换中间部分
- NPOI在.net中的操作Excel
- Eclipse 改动凝视的 date time 日期时间格式,即${date}变量格式
- Java业务原子性的一种实现(key 独占访问)
- phpmyadmin设置编码和字符集gbk或utf8_导入中文乱码解决方法
- Gitlab使用时的一些注意事项
- HDU1262-寻找素数对
- python写商品管理练习
- 解决webdiyer:AspNetPager分页控件在IE模式下点击页号不能翻页的问题。
- python 的序列化和反序列化
- jdbctemplate 获取数据表结构的方法&;注意事项
- ueditor的上传文件漏洞(c#)
- ajax原理以及优缺点(转)
- BSD Socket 通信
- jquery插件开发的demo
- P1155 双栈排序
- window.open居中显示
- centos7中使用LVM管理磁盘和挂载磁盘
热门文章
- 大数据技术生态圈形象比喻(Hadoop、Hive、Spark 关系)
- php能做什么
- Hbase出现ERROR: Can&#39;t get master address from ZooKeeper; znode data == null解决办法
- C#语言中的XmlSerializer类的XmlSerializer.Serialize(Stream,Object)方法举例详解
- Hive入门学习--HIve简介
- HTML编码和CSS编码会遇到的问
- SEO优化-robots.txt解读
- C#的一些小知识
- VS下使用Google Protobuf完成SOCKET通信
- 使用float属性的一些小技巧