func (l *RegistryConfigurationListener) Process(configType *config_center.ConfigChangeEvent) {
	l.events.In() <- configType
}
Process submit the ConfigChangeEvent to the event chan to notify all observer
package zookeeper
import (
	"strings"
	"sync"
)
import (
	gxchan "github.com/dubbogo/gost/container/chan"
	gxzookeeper "github.com/dubbogo/gost/database/kv/zk"
	perrors "github.com/pkg/errors"
)
import (
	"dubbo.apache.org/dubbo-go/v3/common"
	"dubbo.apache.org/dubbo-go/v3/common/logger"
	"dubbo.apache.org/dubbo-go/v3/config_center"
	"dubbo.apache.org/dubbo-go/v3/registry"
	"dubbo.apache.org/dubbo-go/v3/remoting"
)
type RegistryDataListener struct {
	subscribed map[string]config_center.ConfigurationListener
	mutex      sync.Mutex
	closed     bool
}
func NewRegistryDataListener() *RegistryDataListener {
	return &RegistryDataListener{
		subscribed: make(map[string]config_center.ConfigurationListener),
	}
}
func (l *RegistryDataListener) SubscribeURL(url *common.URL, listener config_center.ConfigurationListener) {
	if l.closed {
		return
	}
	l.subscribed[url.ServiceKey()] = listener
}
func (l *RegistryDataListener) UnSubscribeURL(url *common.URL) config_center.ConfigurationListener {
	if l.closed {
		return nil
	}
	listener := l.subscribed[url.ServiceKey()]
	listener.(*RegistryConfigurationListener).Close()
	delete(l.subscribed, url.ServiceKey())
	return listener
}
func (l *RegistryDataListener) DataChange(eventType remoting.Event) bool {
	index := strings.Index(eventType.Path, "/providers/")
	if index == -1 {
		logger.Warnf("Listen with no url, event.path={%v}", eventType.Path)
		return false
	}
	url := eventType.Path[index+len("/providers/"):]
	serviceURL, err := common.NewURL(url)
	if err != nil {
		logger.Errorf("Listen NewURL(r{%s}) = error{%v} eventType.Path={%v}", url, err, eventType.Path)
		return false
	}
	l.mutex.Lock()
	defer l.mutex.Unlock()
	if l.closed {
		return false
	}
	for serviceKey, listener := range l.subscribed {
		if serviceURL.ServiceKey() == serviceKey {
			listener.Process(
				&config_center.ConfigChangeEvent{
					Key:        eventType.Path,
					Value:      serviceURL,
					ConfigType: eventType.Action,
				},
			)
			return true
		}
	}
	return false
}
func (l *RegistryDataListener) Close() {
	l.mutex.Lock()
	defer l.mutex.Unlock()
	l.closed = true
	for _, listener := range l.subscribed {
		listener.(*RegistryConfigurationListener).Close()
	}
}
type RegistryConfigurationListener struct {
	client       *gxzookeeper.ZookeeperClient
	registry     *zkRegistry
	events       *gxchan.UnboundedChan
	isClosed     bool
	close        chan struct{}
	closeOnce    sync.Once
	subscribeURL *common.URL
}
func NewRegistryConfigurationListener(client *gxzookeeper.ZookeeperClient, reg *zkRegistry, conf *common.URL) *RegistryConfigurationListener {
	reg.WaitGroup().Add(1)
	return &RegistryConfigurationListener{
		client:       client,
		registry:     reg,
		events:       gxchan.NewUnboundedChan(32),
		isClosed:     false,
		close:        make(chan struct{}, 1),
		subscribeURL: conf,
	}
}