func (l *EventListener) ListenServiceNodeEventWithPrefix(prefix string, listener ...remoting.DataListener) {
	defer l.wg.Done()
	for {
		wc, done, err := l.client.WatchWithPrefix(prefix)
		if err != nil {
			logger.Warnf("listenDirEvent(key{%s}) = error{%v}", prefix, err)
		}
		select {
		case <-l.client.Done():
			logger.Warnf("kubernetes client stopped")
			return
		case <-done:
			logger.Warnf("kubernetes watcherSet watcher stopped")
			return
		case e, ok := <-wc:
			if !ok {
				logger.Warnf("kubernetes-watcherSet watch-chan closed")
				return
			}
			l.handleEvents(e, listener...)
		}
	}
}
Listen on a set of key with spec prefix
package kubernetes
import (
	"sync"
)
import (
	perrors "github.com/pkg/errors"
)
import (
	"dubbo.apache.org/dubbo-go/v3/common/logger"
	"dubbo.apache.org/dubbo-go/v3/remoting"
)
type EventListener struct {
	client     *Client
	keyMapLock sync.RWMutex
	keyMap     map[string]struct{}
	wg         sync.WaitGroup
}
func NewEventListener(client *Client) *EventListener {
	return &EventListener{
		client: client,
		keyMap: make(map[string]struct{}, 8),
	}
}
func (l *EventListener) ListenServiceNodeEvent(key string, listener ...remoting.DataListener) bool {
	defer l.wg.Done()
	for {
		wc, done, err := l.client.Watch(key)
		if err != nil {
			logger.Warnf("watch exist{key:%s} = error{%v}", key, err)
			return false
		}
		select {
		case <-l.client.Done():
			logger.Warnf("kubernetes client stopped")
			return false
		case <-done:
			logger.Warnf("kubernetes watcherSet watcher stopped")
			return false
		case e, ok := <-wc:
			if !ok {
				logger.Warnf("kubernetes-watcherSet watch-chan closed")
				return false
			}
			if l.handleEvents(e, listener...) {
				return true
			}
		}
	}
}
func (l *EventListener) handleEvents(event *WatcherEvent, listeners ...remoting.DataListener) bool {
	logger.Infof("got a kubernetes-watcherSet event {type: %d, key: %s}", event.EventType, event.Key)
	switch event.EventType {
	case Create:
		for _, listener := range listeners {
			logger.Infof("kubernetes-watcherSet get event (key{%s}) = event{EventNodeDataCreated}", event.Key)
			listener.DataChange(remoting.Event{
				Path:    string(event.Key),
				Action:  remoting.EventTypeAdd,
				Content: string(event.Value),
			})
		}
		return false
	case Update:
		for _, listener := range listeners {
			logger.Infof("kubernetes-watcherSet get event (key{%s}) = event{EventNodeDataChanged}", event.Key)
			listener.DataChange(remoting.Event{
				Path:    string(event.Key),
				Action:  remoting.EventTypeUpdate,
				Content: string(event.Value),
			})
		}
		return false
	case Delete:
		logger.Warnf("kubernetes-watcherSet get event (key{%s}) = event{EventNodeDeleted}", event.Key)
		return true
	default:
		return false
	}
}