godiscovery

README

go-discovery

discovery service for golang

注意：目前对etcd watch方式做下测试，效果不是很理想。本库慎用与生产环境！！！

后续会对 go-discovery 实现机制做回炉改造！！！

功能

• 封装etcd，提供事件触发接口
• 节点自识别自身对外IP
• 节点自动分配可用端口、多节点不会冲突
• 自动分配节点ID、多节点ID不会冲突
• 提供服务列表

例子1

type MyNode struct {
	godiscovery.Node
}

func NewMyNode() *MyNode {
	this := &MyNode{}
	this.Node.Init(this)
	return this
}

func (this *MyNode) OnNodeUpdate(nodeIP string, nodeType int, id uint32, data []byte) {
	fmt.Println("OnNodeUpdate: nodeIP =", nodeIP, "nodeType =", nodeType, "id =", id, "data =", string(data))
}

func (this *MyNode) OnNodeJoin(nodeIP string, nodeType int, id uint32, data []byte) {
	fmt.Println("OnNodeJoin: nodeIP =", nodeIP, "nodeType =", nodeType, "id =", id, "data =", string(data))
}

func (this *MyNode) OnNodeLeave(nodeType int, id uint32) {
	fmt.Println("OnNodeLeave: nodeType =", nodeType, "id =", id)
}

func (this *MyNode) GetPutData() (string, error) {
	return string("here can put overload data!"), nil
}

func main() {

	hosts := ""
	flag.StringVar(&hosts, "hosts", "101.132.47.70:12379,101.132.47.70:22379,101.132.47.70:32379", "etcd hosts")
	whatsmyip := ""
	flag.StringVar(&whatsmyip, "whatsmyip", "101.132.47.70:3000", "whatsmyip host")
	nodeType := 0
	flag.IntVar(&nodeType, "nodeType", 1, "node type")
	watchNodeTypes := ""
	flag.StringVar(&watchNodeTypes, "watchNodeTypes", "1,2,3,4", "watch node type")
	putInterval := int64(0)
	flag.Int64Var(&putInterval, "putInterval", 1, "put interval")

	flag.Parse()

	node := NewMyNode()
	node.OpenByStr(hosts, whatsmyip, nodeType, watchNodeTypes, putInterval)

	for {
		time.Sleep(time.Minute)
	}
}

main函数中参数说明

• hosts

  etcd 地址列表

• whatsmyip

  whatsmyip 地址

• nodeType

  自己节点的类型，非0，则定期(putInterval)调用GetPutData函数，上传数据

• watchNodeTypes

  要监听的节点类型列表。非空，则OnNodeUpdate、OnNodeJoin、OnNodeLeave会被触发

• putInterval

  调用GetPutData函数的频率，单位秒

例子2

获取服务列表

需要import的代码在 github.com/fananchong/go-discovery/serverlist

func NewMyNode() *MyNode {
	this := &MyNode{}
	this.Node = discovery.NewDefaultNode(this)
	return this
}

func (this *MyNode) OnNodeJoin(nodeIP string, nodeType int, id uint32, data []byte) {
	this.Node.OnNodeJoin(nodeIP, nodeType, id, data)

	fmt.Println("print current all node:")
	for t := 1; t <= 4; t++ {
		if lst, ok := this.Servers.GetAll(t); ok {
			for _, info := range lst {
				fmt.Printf("    id:%d type:%d addr:%s\n", info.GetId(), info.GetType(), info.GetExternalIp())
			}
		}
	}
}

func (this *MyNode) OnNodeLeave(nodeType int, id uint32) {
	this.Node.OnNodeLeave(nodeType, id)

	fmt.Println("print current all node:")
	for t := 1; t <= 4; t++ {
		if lst, ok := this.Servers.GetAll(t); ok {
			for _, info := range lst {
				fmt.Printf("    id:%d type:%d addr:%s\n", info.GetId(), info.GetType(), info.GetExternalIp())
			}
		}
	}
}

func main() {

	hosts := ""
	flag.StringVar(&hosts, "hosts", "101.132.47.70:12379,101.132.47.70:22379,101.132.47.70:32379", "etcd hosts")
	whatsmyip := ""
	flag.StringVar(&whatsmyip, "whatsmyip", "101.132.47.70:3000", "whatsmyip host")
	nodeType := 0
	flag.IntVar(&nodeType, "nodeType", 1, "node type")
	watchNodeTypes := ""
	flag.StringVar(&watchNodeTypes, "watchNodeTypes", "1,2,3,4", "watch node type")
	putInterval := int64(0)
	flag.Int64Var(&putInterval, "putInterval", 1, "put interval")

	flag.Parse()

	node := NewMyNode()
	node.OpenByStr(hosts, whatsmyip, nodeType, watchNodeTypes, putInterval)

	for {
		time.Sleep(10 * time.Second)
	}

	node.Close()
}

API使用注意事项

OnNodeUpdate、OnNodeJoin、OnNodeLeave、GetPutData 在内部协程被调用，请注意多协程安全！！！

其他注意事项

经测试，etcd不能频繁put数据，可以通过优化操作：

1. 进程开启时，put一次数据
2. 超过负载或恢复正常时，put一次数据
3. 60s，定期，put数据 (看服务节点数据，来调整)

客户端做轮询方式负载均衡

这样可以极大的减轻etcd的负载

Etcd部署脚本说明

提供2种Docker Swarm方式部署etcd

启动脚本	说明
docker-swarm/install-etcd-static.sh	静态配置方式部署etcd
docker-swarm/install-discovery.etcd.io.sh docker-swarm/install-etcd.sh	etcd发现方式部署etcd

WhatsMyIP部署脚本说明

启动脚本	说明
docker-swarm/install-whatsmyip.sh	启动whatsmyip服务

Documentation

Index

• func SetLogger(log ILogger)
• type DefaultLogger
  • func NewDefaultLogger() *DefaultLogger
  • func (this *DefaultLogger) Error(args ...interface{})
  • func (this *DefaultLogger) Errorf(format string, args ...interface{})
  • func (this *DefaultLogger) Errorln(args ...interface{})
  • func (this *DefaultLogger) Info(args ...interface{})
  • func (this *DefaultLogger) Infof(format string, args ...interface{})
  • func (this *DefaultLogger) Infoln(args ...interface{})
• type ILogger
• type INode
• type IPut
• type IWatch
• type Node
  • func (this *Node) Close()
  • func (this *Node) DisablePort()
  • func (this *Node) GetBase() interface{}
  • func (this *Node) GetClient() *clientv3.Client
  • func (this *Node) GetCtx() context.Context
  • func (this *Node) GetPutData() (string, error)
  • func (this *Node) Id() uint32
  • func (this *Node) Init(inst interface{})
  • func (this *Node) Ip() string
  • func (this *Node) NewNodeId() (uint32, error)
  • func (this *Node) OnNodeJoin(nodeIP string, nodeType int, id uint32, data []byte)
  • func (this *Node) OnNodeLeave(nodeType int, id uint32)
  • func (this *Node) OnNodeUpdate(nodeIP string, nodeType int, id uint32, data []byte)
  • func (this *Node) Open(hosts []string, whatsmyipHost string, nodeType int, watchNodeTypes []int, ...)
  • func (this *Node) OpenByStr(hostsStr string, whatsmyipHost string, nodeType int, watchNodeTypesStr string, ...)
  • func (this *Node) SetId(id uint32)
• type Port
  • func (this *Port) GetPort() uint16
  • func (this *Port) Init(root context.Context, client *clientv3.Client) error
• type Put
  • func (this *Put) Open(root context.Context, nodeType int, putInterval int64) error
• type Watch
  • func (this *Watch) Open(root context.Context, watchNodeTypes []int)

Functions

func SetLogger

func SetLogger(log ILogger)

Types

type DefaultLogger

type DefaultLogger struct {
}

func NewDefaultLogger

func NewDefaultLogger() *DefaultLogger

func (*DefaultLogger) Error

func (this *DefaultLogger) Error(args ...interface{})

func (*DefaultLogger) Errorf

func (this *DefaultLogger) Errorf(format string, args ...interface{})

func (*DefaultLogger) Errorln

func (this *DefaultLogger) Errorln(args ...interface{})

func (*DefaultLogger) Info

func (this *DefaultLogger) Info(args ...interface{})

func (*DefaultLogger) Infof

func (this *DefaultLogger) Infof(format string, args ...interface{})

func (*DefaultLogger) Infoln

func (this *DefaultLogger) Infoln(args ...interface{})

type ILogger

type ILogger interface {
	Info(args ...interface{})
	Infoln(args ...interface{})
	Infof(format string, args ...interface{})
	Error(args ...interface{})
	Errorln(args ...interface{})
	Errorf(format string, args ...interface{})
}

type INode

type INode interface {
	Id() uint32
	SetId(id uint32)
	Init(inst interface{})
	Open(hosts []string, whatsmyip string, nodeType int, watchNodeTypes []int, putInterval int64)
	Close()
	GetClient() *clientv3.Client
	GetBase() interface{}
}

type IPut

type IPut interface {
	INode
	GetPutData() (string, error)
	NewNodeId() (uint32, error)
}

type IWatch

type IWatch interface {
	INode
	OnNodeUpdate(nodeIP string, nodeType int, id uint32, data []byte)
	OnNodeJoin(nodeIP string, nodeType int, id uint32, data []byte)
	OnNodeLeave(nodeType int, id uint32)
}

type Node

type Node struct {
	Watch
	Put
	Port
	// contains filtered or unexported fields
}

func (*Node) Close

func (this *Node) Close()

func (*Node) DisablePort

func (this *Node) DisablePort()

func (*Node) GetBase

func (this *Node) GetBase() interface{}

func (*Node) GetClient

func (this *Node) GetClient() *clientv3.Client

func (*Node) GetCtx

func (this *Node) GetCtx() context.Context

func (*Node) GetPutData

func (this *Node) GetPutData() (string, error)

func (*Node) Id

func (this *Node) Id() uint32

func (*Node) Init

func (this *Node) Init(inst interface{})

func (*Node) Ip

func (this *Node) Ip() string

func (*Node) NewNodeId

func (this *Node) NewNodeId() (uint32, error)

func (*Node) OnNodeJoin

func (this *Node) OnNodeJoin(nodeIP string, nodeType int, id uint32, data []byte)

func (*Node) OnNodeLeave

func (this *Node) OnNodeLeave(nodeType int, id uint32)

func (*Node) OnNodeUpdate

func (this *Node) OnNodeUpdate(nodeIP string, nodeType int, id uint32, data []byte)

子类可以根据需要重载下面的方法

注意 OnNodeUpdate、OnNodeJoin、OnNodeLeave、GetPutData 在内部协程被调用，请注意多协程安全！！！

func (*Node) Open

func (this *Node) Open(hosts []string, whatsmyipHost string, nodeType int, watchNodeTypes []int, putInterval int64)

func (*Node) OpenByStr

func (this *Node) OpenByStr(hostsStr string, whatsmyipHost string, nodeType int, watchNodeTypesStr string, putInterval int64)

func (*Node) SetId

func (this *Node) SetId(id uint32)

type Port

type Port struct {
	// contains filtered or unexported fields
}

func (*Port) GetPort

func (this *Port) GetPort() uint16

func (*Port) Init

func (this *Port) Init(root context.Context, client *clientv3.Client) error

type Put

type Put struct {
	Derived IPut

	Id uint32
	// contains filtered or unexported fields
}

func (*Put) Open

func (this *Put) Open(root context.Context, nodeType int, putInterval int64) error

type Watch

type Watch struct {
	Derived IWatch
	// contains filtered or unexported fields
}

func (*Watch) Open

func (this *Watch) Open(root context.Context, watchNodeTypes []int)