func (r *kubernetesRegistry) InitListeners() {
	r.listener = kubernetes.NewEventListener(r.client)
	r.configListener = NewConfigurationListener(r)
	r.dataListener = NewRegistryDataListener(r.configListener)
}
InitListeners init listeners of kubernetes registry center
package kubernetes
import (
	"fmt"
	"path"
	"sync"
	"time"
)
import (
	gxtime "github.com/dubbogo/gost/time"
	perrors "github.com/pkg/errors"
	v1 "k8s.io/api/core/v1"
)
import (
	"dubbo.apache.org/dubbo-go/v3/common"
	"dubbo.apache.org/dubbo-go/v3/common/constant"
	"dubbo.apache.org/dubbo-go/v3/common/extension"
	"dubbo.apache.org/dubbo-go/v3/common/logger"
	"dubbo.apache.org/dubbo-go/v3/registry"
	"dubbo.apache.org/dubbo-go/v3/remoting/kubernetes"
)
const (
	Name         = "kubernetes"
	ConnDelay    = 3
	MaxFailTimes = 15
)
func init() {
	extension.SetRegistry(Name, newKubernetesRegistry)
}
type kubernetesRegistry struct {
	registry.BaseRegistry
	cltLock        sync.RWMutex
	client         *kubernetes.Client
	listenerLock   sync.Mutex
	listener       *kubernetes.EventListener
	dataListener   *dataListener
	configListener *configurationListener
}
func (r *kubernetesRegistry) Client() *kubernetes.Client {
	r.cltLock.RLock()
	client := r.client
	r.cltLock.RUnlock()
	return client
}
func (r *kubernetesRegistry) SetClient(client *kubernetes.Client) {
	r.cltLock.Lock()
	r.client = client
	r.cltLock.Unlock()
}
func (r *kubernetesRegistry) CloseAndNilClient() {
	r.client.Close()
	r.client = nil
}
func (r *kubernetesRegistry) CloseListener() {
	r.cltLock.Lock()
	l := r.configListener
	r.cltLock.Unlock()
	if l != nil {
		l.Close()
	}
	r.configListener = nil
}
func (r *kubernetesRegistry) CreatePath(k string) error {
	if err := r.client.Create(k, ""); err != nil {
		return perrors.WithMessagef(err, "create path %s in kubernetes", k)
	}
	return nil
}
func (r *kubernetesRegistry) DoRegister(root string, node string) error {
	return r.client.Create(path.Join(root, node), "")
}
func (r *kubernetesRegistry) DoUnregister(root string, node string) error {
	return perrors.New("DoUnregister is not support in kubernetesRegistry")
}
func (r *kubernetesRegistry) DoSubscribe(svc *common.URL) (registry.Listener, error) {
	var configListener *configurationListener
	r.listenerLock.Lock()
	configListener = r.configListener
	r.listenerLock.Unlock()
	if r.listener == nil {
		r.cltLock.Lock()
		client := r.client
		r.cltLock.Unlock()
		if client == nil {
			return nil, perrors.New("kubernetes client broken")
		}
		r.listenerLock.Lock()
		if r.listener == nil {
			r.listener = kubernetes.NewEventListener(r.client)
		}
		r.listenerLock.Unlock()
	}
	r.dataListener.AddInterestedURL(svc)
	go r.listener.ListenServiceEvent(fmt.Sprintf("/dubbo/%s/"+constant.DEFAULT_CATEGORY, svc.Service()), r.dataListener)
	return configListener, nil
}
func (r *kubernetesRegistry) DoUnsubscribe(conf *common.URL) (registry.Listener, error) {
	return nil, perrors.New("DoUnsubscribe is not support in kubernetesRegistry")
}