根据watch信息拉取指定应用、服务名的实例信息

Author: chengyouling

clients/service-center-client/src/main/java/org/apache/servicecomb/service/center/client/DiscoveryEvents.java

@@ -20,6 +20,11 @@
 import java.util.List;
 
 import org.apache.servicecomb.service.center.client.model.MicroserviceInstance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 
 public abstract class DiscoveryEvents {
   public static class InstanceChangedEvent extends DiscoveryEvents {
@@ -52,6 +57,46 @@ public List<MicroserviceInstance> getInstances() {
    * internal events to ask for a immediate instance pull
    */
   public static class PullInstanceEvent extends DiscoveryEvents {
+    private static final Logger LOGGER = LoggerFactory.getLogger(PullInstanceEvent.class);
+
+    private final String appId;
+
+    private final String serviceName;
+
+    public PullInstanceEvent(String message) {
+      JsonNode messageNode = parseJsonString(message);
+      this.appId = getContextFromNode(messageNode, "appId");
+      this.serviceName = getContextFromNode(messageNode, "serviceName");
+    }
+
+    public String getAppId() {
+      return appId;
+    }
 
+    public String getServiceName() {
+      return serviceName;
+    }
+
+    private JsonNode parseJsonString(String message) {
+      try {
+        ObjectMapper objectMapper = new ObjectMapper();
+        return objectMapper.readTree(message);
+      } catch (Exception e) {
+        LOGGER.error("parse message [{}] failed!", message, e);
+        return null;
+      }
+    }
+
+    private String getContextFromNode(JsonNode messageNode, String itemKey) {
+      if (messageNode == null) {
+        return "";
+      }
+      try {
+        return messageNode.get("key").get(itemKey).asText();
+      } catch (Exception e) {
+        LOGGER.error("get [{}] context from node [{}] failed!", itemKey, e);
+        return "";
+      }
+    }
   }
 }

clients/service-center-client/src/main/java/org/apache/servicecomb/service/center/client/ServiceCenterClient.java

@@ -237,8 +237,7 @@ public RegisteredMicroserviceInstanceResponse registerMicroserviceInstance(Micro
 
   @Override
   public FindMicroserviceInstancesResponse findMicroserviceInstance(String consumerId, String appId, String serviceName,
-      String versionRule,
-      String revision) {
+      String versionRule, String revision) {
     try {
       Map<String, String> headers = new HashMap<>();
       headers.put("X-ConsumerId", consumerId);

clients/service-center-client/src/main/java/org/apache/servicecomb/service/center/client/ServiceCenterDiscovery.java

@@ -24,6 +24,7 @@
 import java.util.Random;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.servicecomb.http.client.task.AbstractTask;
 import org.apache.servicecomb.http.client.task.Task;
 import org.apache.servicecomb.service.center.client.DiscoveryEvents.InstanceChangedEvent;
@@ -151,7 +152,45 @@ public void onPullInstanceEvent(PullInstanceEvent event) {
       return;
     }
     pullInstanceTaskOnceInProgress = true;
-    startTask(new PullInstanceOnceTask());
+    if (StringUtils.isEmpty(event.getAppId()) || StringUtils.isEmpty(event.getServiceName())) {
+      // If the application or service name cannot be resolved, pulled all services.
+      startTask(new PullInstanceOnceTask());
+      return;
+    }
+    try {
+      pullTargetService(event);
+    } finally {
+      pullInstanceTaskOnceInProgress = false;
+    }
+  }
+
+  private void pullTargetService(PullInstanceEvent event) {
+    SubscriptionKey currentKey = new SubscriptionKey(event.getAppId(), event.getServiceName());
+    if (instancesCache.get(currentKey) == null) {
+      // No pull during the service startup phase.
+      return;
+    }
+    if (LOGGER.isDebugEnabled()) {
+      LOGGER.debug("pull [{}#{}] instances from service center", event.getAppId(), event.getServiceName());
+    }
+    String originRev = instancesCache.get(currentKey).revision;
+    for (int i = 1; i <= 3; i++) {
+      try {
+        pullInstance(currentKey, instancesCache.get(currentKey), true);
+        String currentRev = instancesCache.get(currentKey).revision;
+        if (LOGGER.isDebugEnabled()) {
+          LOGGER.debug("current revision: [{}], origin revision: [{}]", currentRev, originRev);
+        }
+        if (!originRev.equals(currentRev)) {
+          break;
+        }
+        int positive = random.nextInt(300);
+        int sign = random.nextBoolean() ? 1 : -1;
+        Thread.sleep(i * 1000 + sign * positive);
+      } catch (Exception e) {
+        LOGGER.error("pull [{}#{}] instances failed.", event.getAppId(), event.getServiceName(), e);
+      }
+    }
   }
 
   private void pullInstance(SubscriptionKey k, SubscriptionValue v, boolean sendChangedEvent) {
@@ -250,7 +289,7 @@ private static String instanceToString(List<MicroserviceInstance> instances) {
         sb.append(endpoint.length() > 64 ? endpoint.substring(0, 64) : endpoint);
         sb.append("|");
       }
-      sb.append(instance.getServiceName());
+      sb.append(instance.getStatus());
       sb.append("|");
     }
     sb.append("#");

clients/service-center-client/src/main/java/org/apache/servicecomb/service/center/client/ServiceCenterOperation.java

@@ -114,8 +114,7 @@ public interface ServiceCenterOperation {
    * @throws OperationException If some problems happened to contact service center or non http 200 returned.n
    */
   FindMicroserviceInstancesResponse findMicroserviceInstance(String consumerId, String appId, String serviceName,
-      String versionRule,
-      String revision);
+      String versionRule, String revision);
 
   /**
    * Delete a microservice instance

clients/service-center-client/src/main/java/org/apache/servicecomb/service/center/client/ServiceCenterWatch.java

@@ -177,7 +177,7 @@ private void backOff() {
   @Override
   public void onMessage(String s) {
     LOGGER.info("web socket receive message [{}], start query instance", s);
-    this.eventBus.post(new PullInstanceEvent());
+    this.eventBus.post(new PullInstanceEvent(s));
   }
 
   @Override