AlexStocks commented on code in PR #3634:
URL: https://github.com/apache/dubbo-go/pull/3634#discussion_r3749849916
##########
registry/servicediscovery/service_discovery_registry.go:
##########
@@ -576,28 +582,185 @@ func (s *serviceDiscoveryRegistry)
getServiceListener(serviceNamesKey string) re
return s.serviceListeners[serviceNamesKey]
}
+// protocolServiceKeyOf builds the subscriber key SubscribeURL registers and
+// UnSubscribe removes: consumers default to the "tri" protocol, other
+// protocols need to be specified on the reference/consumer explicitly.
+func protocolServiceKeyOf(url *common.URL) string {
+ protocol := constant.TriProtocol
+ if url.Protocol != "" {
+ protocol = url.Protocol
+ }
+ return url.ServiceKey() + ":" + protocol
+}
+
+// loadLatestInstances pushes the current registry snapshot for every
subscribed
+// application into the listener. It runs outside s.lock: GetInstances and
+// OnEvent may perform external RPC / metadata-report calls.
+func (s *serviceDiscoveryRegistry) loadLatestInstances(listener
registry.ServiceInstancesChangedListener) {
+ for _, serviceNameTmp := range listener.GetServiceNames().Values() {
+ serviceName := serviceNameTmp.(string)
+ instances := s.serviceDiscovery.GetInstances(serviceName)
+ logger.Infof("[Registry][ServiceDiscovery] synchronized
instance notification on application %s subscription, instance list size %d",
serviceName, len(instances))
+ if err :=
listener.OnEvent(®istry.ServiceInstancesChangedEvent{
+ ServiceName: serviceName,
+ Instances: instances,
+ }); err != nil {
+ logger.Warnf("[Registry][ServiceDiscovery]
ServiceInstancesChangedListenerImpl handle error, err=%v", err)
+ }
+ }
+}
+
// subscribeAndNotify registers the notify callback and asynchronously wires
the
-// listener into the service discovery so the caller does not block on it.
+// listener into the service discovery so the caller does not block on it. A
+// failed AddListener is retried in the background with backoff; without the
+// retry a transient registry error would leave the consumer permanently stale
+// (issue #3624).
func (s *serviceDiscoveryRegistry) subscribeAndNotify(url *common.URL,
serviceNamesKey, protocolServiceKey string,
listener registry.ServiceInstancesChangedListener, notify
registry.NotifyListener,
) {
listener.AddListenerAndNotify(protocolServiceKey, notify)
- event :=
metricsMetadata.NewMetadataMetricTimeEvent(metricsMetadata.SubscribeServiceRt)
logger.Infof("[Registry][ServiceDiscovery] start subscribing to
registry for applications=%s with a new go routine", serviceNamesKey)
go func() {
- err := s.serviceDiscovery.AddListener(listener)
- event.Succ = err != nil
- event.End = time.Now()
- event.Attachment[constant.InterfaceKey] = url.Interface()
- metrics.Publish(event)
- metrics.Publish(metricsRegistry.NewServerSubscribeEvent(err ==
nil))
- if err != nil {
+ if err := s.addInstanceListener(url, listener); err != nil {
Review Comment:
[P1] 后续成功的 `AddListener` 不会撤销同一 `serviceNamesKey` 已挂起的重试。首个订阅失败后，如果 timer
触发前另一个 `SubscribeURL` 成功，这个 goroutine 的成功分支什么都不做，旧 timer 仍会再次调用
`AddListener`；当前 Head 的公开入口探针得到 2 次必要调用后仍出现第 3 次。Polaris 的 `AddSubscriber` 和
ZooKeeper 的 `ListenServiceEvent` 会重复注册回调/监听，不能依赖幂等。成功时应在确认 listener/state 匹配后取消
pending retry，并补“先失败、后续订阅成功、等待超过 backoff 仍只有两次调用”的测试。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
