func NewGroup(name, path string, node *JobNode) (group *Group) {
	group = &Group{
		name:      name,
		path:      path,
		node:      node,
		watchPath: fmt.Sprintf(ClientPath, name),
		clients:   make(map[string]*Client),
		lk:        &sync.RWMutex{},
	}
	go group.watchClientPath()
	go group.loopLoadAllClient()
	return
}
create a new group
package forest
import (
	"context"
	"errors"
	"fmt"
	"github.com/coreos/etcd/clientv3"
	"github.com/labstack/gommon/log"
	"math/rand"
	"sync"
	"time"
)
const (
	GroupConfPath = "/forest/server/group/"
	ClientPath    = "/forest/client/%s/clients/"
)
type JobGroupManager struct {
	node   *JobNode
	groups map[string]*Group
	lk     *sync.RWMutex
}
func NewJobGroupManager(node *JobNode) (mgr *JobGroupManager) {
	mgr = &JobGroupManager{
		node:   node,
		groups: make(map[string]*Group),
		lk:     &sync.RWMutex{},
	}
	go mgr.watchGroupPath()
	go mgr.loopLoadGroups()
	return
}
func (mgr *JobGroupManager) watchGroupPath() {
	keyChangeEventResponse := mgr.node.etcd.WatchWithPrefixKey(GroupConfPath)
	for ch := range keyChangeEventResponse.Event {
		mgr.handleGroupChangeEvent(ch)
	}
}
func (mgr *JobGroupManager) loopLoadGroups() {
RETRY:
	var (
		keys   [][]byte
		values [][]byte
		err    error
	)
	if keys, values, err = mgr.node.etcd.GetWithPrefixKey(GroupConfPath); err != nil {
		goto RETRY
	}
	if len(keys) == 0 {
		return
	}
	for i := 0; i < len(keys); i++ {
		path := string(keys[i])
		groupConf, err := UParkGroupConf(values[i])
		if err != nil {
			log.Warnf("upark the group conf error:%#v", err)
			continue
		}
		mgr.addGroup(groupConf.Name, path)
	}
}
func (mgr *JobGroupManager) addGroup(name, path string) {
	mgr.lk.Lock()
	defer mgr.lk.Unlock()
	if _, ok := mgr.groups[path]; ok {
		return
	}
	group := NewGroup(name, path, mgr.node)
	mgr.groups[path] = group
	log.Infof("add a new group:%s,for path:%s", name, path)
}
func (mgr *JobGroupManager) deleteGroup(path string) {
	var (
		group *Group
		ok    bool
	)
	mgr.lk.Lock()
	defer mgr.lk.Unlock()
	if group, ok = mgr.groups[path]; ok {
		return
	}
	_ = group.watcher.Close()
	group.cancelFunc()
	delete(mgr.groups, path)
	log.Infof("delete a  group:%s,for path:%s", group.name, path)
}
func (mgr *JobGroupManager) handleGroupChangeEvent(changeEvent *KeyChangeEvent) {
	switch changeEvent.Type {
	case KeyCreateChangeEvent:
		mgr.handleGroupCreateEvent(changeEvent)
	case KeyUpdateChangeEvent:
	case KeyDeleteChangeEvent:
		mgr.handleGroupDeleteEvent(changeEvent)
	}
}
func (mgr *JobGroupManager) handleGroupCreateEvent(changeEvent *KeyChangeEvent) {
	groupConf, err := UParkGroupConf(changeEvent.Value)
	if err != nil {
		log.Warnf("upark the group conf error:%#v", err)
		return
	}
	path := changeEvent.Key
	mgr.addGroup(groupConf.Name, path)
}
func (mgr *JobGroupManager) handleGroupDeleteEvent(changeEvent *KeyChangeEvent) {
	path := changeEvent.Key
	mgr.deleteGroup(path)
}
func (mgr *JobGroupManager) selectClient(name string) (client *Client, err error) {
	var (
		group *Group
		ok    bool
	)
	if group, ok = mgr.groups[GroupConfPath+name]; !ok {
		err = errors.New(fmt.Sprintf("the group:%s not found", name))
		return
	}
	return group.selectClient()
}
type Group struct {
	path       string
	name       string
	node       *JobNode
	watchPath  string
	clients    map[string]*Client
	watcher    clientv3.Watcher
	cancelFunc context.CancelFunc
	lk         *sync.RWMutex
}