1等待时间,10ms 优化

2配置 ，url 不需要。引用公共的

matrix-gracefulshutdown/src/main/java/com/secoo/mall/dubbo/monitor/config/ConfigCenter.java

@@ -36,31 +36,18 @@ public class ConfigCenter {
 
     public static final Logger logger = LoggerFactory.getLogger(ConfigCenter.class);
 
-    @Value("${admin.registry.address:}")
+    @Value("${dubbo.registry.address:}")
     public String registryAddress;
-
-    @Value("${admin.registry.group:dubbo}")
+    @Value("${dubbo.protocol.name}")
     public String registryGroup;
-    @Value("${admin.config-center.group:dubbo}")
-    public String configCenterGroup;
-
-    @Value("${admin.config-center.username:}")
+    @Value("${dubbo.registry.username:}")
     public String username;
-    @Value("${admin.config-center.password:}")
+    @Value("${dubbo.registry.password:}")
     public String password;
-    @Value("${admin.patch:}")
+    @Value("${dubbo.monitorpatch:}")
     public String patch;
-    public URL configCenterUrl;
     public URL registryUrl;
 
-    public String getRegistryGroup() {
-        return registryGroup;
-    }
-
-    public void setRegistryGroup(String registryGroup) {
-        this.registryGroup = registryGroup;
-    }
-
     public URL getRegistryUrl() {
         return registryUrl;
     }

matrix-gracefulshutdown/src/main/java/com/secoo/mall/dubbo/monitor/dubbo/service/RegistryServerSync.java

@@ -43,7 +43,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicLong;
 
-public class RegistryServerSync implements InitializingBean, DisposableBean, NotifyListener, CommandLineRunner {
+public class RegistryServerSync extends Observable implements InitializingBean, DisposableBean, NotifyListener, CommandLineRunner  {
     public RegistryServerSync(Registry registry) {
          this.registry=registry;
     }
@@ -73,7 +73,7 @@ public class RegistryServerSync implements InitializingBean, DisposableBean, Not
      * ConcurrentMap<category, ConcurrentMap<servicename, Map<MD5, URL>>>
      * registryCache
      */
-    private final ConcurrentMap<String, ConcurrentMap<String, Map<String, URL>>> registryCache = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, ConcurrentMap<String, Map<String, URL>>>   registryCache = new ConcurrentHashMap<>();
  //  @Autowired
 
     public Registry registry;
@@ -83,6 +83,16 @@ public class RegistryServerSync implements InitializingBean, DisposableBean, Not
         return registryCache;
     }
 
+    volatile boolean  signal=false;
+
+    public boolean isSignal() {
+        return signal;
+    }
+
+    public void setSignal(boolean signal) {
+        this.signal = signal;
+    }
+
     @Override
     public void afterPropertiesSet() throws Exception {
         logger.info("Init Dubbo Admin Sync Cache...");
@@ -119,15 +129,17 @@ public class RegistryServerSync implements InitializingBean, DisposableBean, Not
                     String group = url.getParameter(Constants.GROUP_KEY);
                     String version = url.getParameter(Constants.VERSION_KEY);
                     // NOTE: group and version in empty protocol is *
-                    if (!Constants.ANY_VALUE.equals(group) && !Constants.ANY_VALUE.equals(version)) {
-                        services.remove(url.getServiceKey());
+                    if (!Constants.ANY_VALUE.equals(group) && !Constants.ANY_VALUE.equals(version)) {//移除带版本的key
+                      //  services.remove(url.getServiceKey());
+                        removeAndNotice(services,url.getServiceKey());
                     } else {
                         for (Map.Entry<String, Map<String, URL>> serviceEntry : services.entrySet()) {
                             String service = serviceEntry.getKey();
                             if (Tool.getInterface(service).equals(url.getServiceInterface())
                                     && (Constants.ANY_VALUE.equals(group) || StringUtils.isEquals(group, Tool.getGroup(service)))
                                     && (Constants.ANY_VALUE.equals(version) || StringUtils.isEquals(version, Tool.getVersion(service)))) {
-                                services.remove(service);//版本 和分组匹配就干掉？
+                                services.remove(service);//如果版本和分组匹配就干掉
+                                removeAndNotice(services,service);
                             }
                         }
                     }
@@ -141,7 +153,7 @@ public class RegistryServerSync implements InitializingBean, DisposableBean, Not
                     services = new HashMap<>();
                     categories.put(category, services);
                 }
-                String service = url.getServiceKey();
+                String service = url.getServiceKey();//带版本
                 Map<String, URL> ids = services.get(service);
                 if (ids == null) {
                     ids = new HashMap<>();
@@ -171,7 +183,8 @@ public class RegistryServerSync implements InitializingBean, DisposableBean, Not
                 Set<String> keys = new HashSet<String>(services.keySet());
                 for (String key : keys) { //接口
                     if (Tool.getInterface(key).equals(interfaceName) && !categoryEntry.getValue().entrySet().contains(key)) {
-                        services.remove(key);
+                       // services.remove(key);
+                        removeAndNotice(services,key);
                     }
                 }
             }
@@ -219,5 +232,13 @@ public class RegistryServerSync implements InitializingBean, DisposableBean, Not
         }
 
     }
+    public void removeAndNotice(ConcurrentMap<String, Map<String, URL>> services,String serviceKey){
+        services.remove(serviceKey);
+        if(signal) {
+            this.setChanged();
+            this.notifyObservers(serviceKey);
+        }
+
+    }
 }
 

matrix-gracefulshutdown/src/main/java/com/secoo/mall/dubbo/service/GracefullyShoutDown.java

@@ -7,6 +7,7 @@ import com.secoo.mall.common.util.date.DateUtil;
 import com.secoo.mall.common.util.log.LoggerUtil;
 import com.secoo.mall.common.core.bean.gracefulshowtdownBean.ExecutorDetail;
 import com.secoo.mall.common.core.bean.gracefulshowtdownBean.ExecutorDetails;
+import com.secoo.mall.dubbo.monitor.dubbo.service.RegistryServerSync;
 import com.secoo.mall.dubbo.service.impl.TomcatConnectShoutDownHock;
 import org.apache.dubbo.common.utils.NetUtils;
 import org.apache.dubbo.config.DubboShutdownHook;
@@ -34,6 +35,8 @@ public class GracefullyShoutDown implements CommandLineRunner, ApplicationListen
     private ApplicationContext context;
   //  private  ProviderService providerService ;
 
+    RegistryServerSync registryServerSync;
+
     UpDatas dubboUpdata;
 
 //    public  GracefullyShoutDown(TomcatConnectShoutDownHock tomcatConnectShoutDownHock,ProviderService providerService,UpDatas dubboUpdata){
@@ -62,28 +65,29 @@ public class GracefullyShoutDown implements CommandLineRunner, ApplicationListen
         }
 
 
+
         //test-------------------------------------> begin
-//        ExecutorDetails result = new ExecutorDetails();
-//        result.setCode(0);
-//        List<ExecutorDetail> details = new ArrayList<ExecutorDetail>();
-//        result.setT(details);
-//        String ip = NetUtils.getIpByHost(NetUtils.getLocalAddress().getHostName());
-//        String name = ApplicationModel.getApplication();
-//        result.setIp(ip);
-//        result.setName(name);
-//
-//            Map<String, StopService> map = this.context.getBeansOfType(StopService.class);
-//            if (map != null && map.size() > 0) {
-//                Set<StopService> ts = new TreeSet<StopService>();
-//                for (StopService service : map.values()) {
-//                    if (!(service instanceof TomcatConnectShoutDownHock)) {
-//                        ts.add(service);
-//                    }
-//                }
-//                //按定义顺序执行
-//                Iterator<StopService> it = ts.iterator();
-//                while (it.hasNext()) {
-//                    StopService service = (StopService) it.next();
+        ExecutorDetails result = new ExecutorDetails();
+        result.setCode(0);
+        List<ExecutorDetail> details = new ArrayList<ExecutorDetail>();
+        result.setT(details);
+        String ip = NetUtils.getIpByHost(NetUtils.getLocalAddress().getHostName());
+        String name = ApplicationModel.getApplication();
+        result.setIp(ip);
+        result.setName(name);
+
+            Map<String, StopService> map = this.context.getBeansOfType(StopService.class);
+            if (map != null && map.size() > 0) {
+                Set<StopService> ts = new TreeSet<StopService>();
+                for (StopService service : map.values()) {
+                    if (!(service instanceof TomcatConnectShoutDownHock)) {
+                        ts.add(service);
+                    }
+                }
+                //按定义顺序执行
+                Iterator<StopService> it = ts.iterator();
+                while (it.hasNext()) {
+                    StopService service = (StopService) it.next();
 //                    ExecutorDetail one = (ExecutorDetail) service.stop();
 //                    if (one != null) {
 //                        details.add(one);
@@ -91,21 +95,17 @@ public class GracefullyShoutDown implements CommandLineRunner, ApplicationListen
 //                            result.setCode(-1);
 //                        }
 //                    }
-//
-//                }
-//            } catch (Exception e) {
-//                LoggerUtil.error("matrix.GracefullyShoutDown.error", e);
-//            }
-//            try {
-//                dubboUpdata.upData(result);
-//            } catch (Exception e) {
-//                LoggerUtil.error("matrix.GracefullyShoutDown update .error data:" + JSON.toJSONString(result), e);
-//            }
-   //         }
 
+                }
+                try {
+                    dubboUpdata.upData(result);
+                } catch (Exception e) {
+                    LoggerUtil.error("matrix.GracefullyShoutDown update .error data:" + JSON.toJSONString(result), e);
+                }
 
-            //test------------------------------------->
 
+                //test------------------------------------->
+            }
 
     }
 

matrix-gracefulshutdown/src/main/java/com/secoo/mall/dubbo/service/impl/DubboCustomerDownHock.java

@@ -6,6 +6,7 @@ import com.secoo.mall.common.util.date.DateUtil;
 import com.secoo.mall.common.util.log.LoggerUtil;
 import com.secoo.mall.common.util.string.StringUtil;
 import com.secoo.mall.dubbo.monitor.dubbo.service.ProviderService;
+import com.secoo.mall.dubbo.monitor.dubbo.service.RegistryServerSync;
 import com.secoo.mall.dubbo.monitor.utils.Stack;
 import com.secoo.mall.dubbo.service.AbstractStop;
 import org.apache.dubbo.common.URL;
@@ -24,28 +25,43 @@ import org.slf4j.LoggerFactory;
 import java.lang.reflect.Field;
 import java.text.SimpleDateFormat;
 import java.util.*;
+import java.util.concurrent.*;
 
 
-public class DubboCustomerDownHock extends AbstractStop {
-    //@Autowired
+public class DubboCustomerDownHock extends AbstractStop implements Observer {
     private ProviderService providerService;
 
+    RegistryServerSync registryServerSync;
+
+    public RegistryServerSync getRegistryServerSync() {
+        return registryServerSync;
+    }
+
+    public void setRegistryServerSync(RegistryServerSync registryServerSync) {
+        this.registryServerSync = registryServerSync;
+    }
+
     public DubboCustomerDownHock(ProviderService providerService) {
         this.providerService = providerService;
     }
 
     private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
+    private volatile boolean check = false;
+    Map<String, Boolean> checkContent = new HashMap<String, Boolean>();//string 是key boolean 代表是否查找了
+    String ip;
+    String name;
+
     @Override
     public ExecutorDetail stop() {
         ExecutorDetail detail = new ExecutorDetail();
         detail.setBeginTime(DateUtil.getDateTime());
-        detail.setServiceName("dubboDownHock1025");
+        detail.setServiceName("dubboDownHock1038");
         List<String> str = new ArrayList<String>();
         detail.setDetail(str);
         detail.setCode(0);
-        String ip = NetUtils.getIpByHost(NetUtils.getLocalAddress().getHostName());
-        String name = ApplicationModel.getApplication();
+        ip = NetUtils.getIpByHost(NetUtils.getLocalAddress().getHostName());
+        name = ApplicationModel.getApplication();
         try {
 
             Collection<Registry> registries = AbstractRegistryFactory.getRegistries();
@@ -69,11 +85,15 @@ public class DubboCustomerDownHock extends AbstractStop {
                                     str.add("provider url executer unregister before:" + value.getServiceInterface() + " time:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(new Date()) +StringUtil.line());
                                    // throw new IllegalStateException("failed to tet");
                                     zookperRegster.unregister((URL) value);
+                                    registryServerSync.setSignal(true);
+
                                     LoggerUtil.info("provider url unregister success  :" + value.getServiceInterface() + " time:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(new Date()) +StringUtil.line());
                                     str.add("delete  provider url success:" + value.getServiceInterface() + " time:" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS").format(new Date()) +StringUtil.line());
                                 } catch (Exception e) {
                                     LoggerUtil.error("matirx-monitor  deleteService error:", e);
                                     str.add("matirx-monitor  delete Service error service:"+value.getServiceInterface() + StringUtil.line() +Stack.errInfo(e)+StringUtil.line());
+                                }finally {
+                                    registryServerSync.setSignal(true);
                                 }
                             }
                         }else{
@@ -88,31 +108,17 @@ public class DubboCustomerDownHock extends AbstractStop {
             }
             //step3 check检查
             if(checks!=null&&checks.size()>0) {
-                //TODO 监听hashmap的数据变化
                 CuratorZookeeperClient client = (CuratorZookeeperClient) getFieldValueByFieldName("zkClient", zookperRegster);
-                int sessionExpireMs = client.getUrl().getParameter("zk.session.expire", 10000);
-                Thread.sleep(sessionExpireMs);//zookper 目前最大超时时间
-
-                   for( URL url :checks) {
-                    if (StringUtil.isNotEmpty(url.getAddress()) && StringUtil.isNotEmpty(name)) {
-                        List<String> check = null;
-
-                        try {
-                         //   throw new IllegalStateException("failed to tet");
-                            check = providerService.findServicesByAddressAndName(url.getAddress(), name);
-                        } catch (Exception e) {
-                            detail.setCode(-1);
-                            str.add(" error :matirx-monitor  checko Dubbo service  " + StringUtil.line()+ Stack.errInfo(e) + StringUtil.line());
-                        }
-                        if (check != null && check.size() > 0) {
-                            str.add("warn:matrix-monitor check service failure " + url.getServiceKey() + "  reason zk notice failed " + StringUtil.line());
-                            detail.setCode(-1);
-                        }else{
-                            if(check!=null&&check.size()==0) {
-                                str.add("matrix-monitor dubbo check service ok " + url.getServiceKey() + StringUtil.line());
-                            }
-                        }
+                int sessionExpireMs = client.getUrl().getParameter("zk.session.expire", 6000);
+                long startTime=System.currentTimeMillis();   //获取开始时间
+                detail = checkNoticed(detail, str, checks, sessionExpireMs);//通过通知进行检查
+                long endTime=System.currentTimeMillis(); //获取结束时间
+                System.out.println("程序运行时间------------>： "+(endTime-startTime)+"ms");
 
+                if (checks.size() > 0) {   // 处理 没有接到通知的场景
+                    logger.info("NNNNNNNNNNNNNNNNNNNNNNNNOOOOOOOOOOOO!"+JSON.toJSONString(checks));
+                    for (URL check : checks) {
+                        checkService(check, checks, detail, str);
                     }
                 }
             }else{
@@ -173,4 +179,108 @@ public class DubboCustomerDownHock extends AbstractStop {
     }
 
 
+    @Override
+    public void update(Observable o, Object arg) {
+        logger.info("result:" + arg);
+        synchronized (checkContent) {
+            check = true;
+            checkContent.put((String) arg, false);
+        }
+    }
+
+    public ExecutorDetail checkNoticed(ExecutorDetail detail, List<String> str, List<URL> checks, int time) {
+        ExecutorService es = Executors.newSingleThreadExecutor();
+        Future<ExecutorDetail> future = es.submit(new CheckDubbo(detail, str, checks, time));
+        try {
+            return future.get(time, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            logger.error("matrix dubboshutdown check error", e);
+        } catch (ExecutionException e) {
+            logger.error("matrix dubboshutdown check error", e);
+        } catch (TimeoutException e) {
+            logger.error("matrix dubboshutdown check error", e);
+        } finally {
+            if (future.isDone()) {
+                //TODO 结束线程，如何结束？
+            }
+            es.shutdownNow();
+        }
+        return detail;
+    }
+
+    //检查服务逻辑
+    public List<String> checkService(URL url, List<URL> urls, ExecutorDetail detail, List<String> content) {
+        List<String> result = null;
+        try {
+            //   throw new IllegalStateException("failed to tet");
+            result = providerService.findServicesByAddressAndName(url.getAddress(), name);
+        } catch (Exception e) {
+            urls.remove(url);
+            detail.setCode(-1);
+            content.add(" error :matirx-monitor  checko Dubbo service  " + StringUtil.line() + Stack.errInfo(e) + StringUtil.line());
+        }
+        if (result != null && result.size() > 0) {
+            content.add("warn:matrix-monitor check service failure " + url.getServiceKey() + "  reason zk notice failed " + StringUtil.line());
+            detail.setCode(-1);
+        } else {
+            if (result != null && result.size() == 0) {
+                content.add("matrix-monitor dubbo check service ok " + url.getServiceKey() + StringUtil.line());
+            }
+        }
+        logger.info("checkServicecheckServicecheckServicecheckServicecheckServicecheckServicecheckServicecheckService!!!!!");
+        return result;
+    }
+
+    private class CheckDubbo implements Callable<ExecutorDetail> {
+        ExecutorDetail detail;
+        List<String> str;
+        List<URL> checks;
+        int time;
+
+        public CheckDubbo(ExecutorDetail detail, List<String> str, List<URL> checks,int time) {
+            this.detail = detail;
+            this.str = str;
+            this.checks = checks;
+            this.time=time;
+        }
+
+        /**
+         * Computes a result, or throws an exception if unable to do so.
+         *
+         * @return computed result
+         * @throws Exception if unable to compute a result
+         */
+        @Override
+        public ExecutorDetail call() throws Exception {
+            long beginTime=System.currentTimeMillis();
+            //集合转换方便查找
+            Map<String, URL> allService = new HashMap<String, URL>();
+            for (URL url : checks) {
+                allService.put(url.getServiceKey(), url);
+            }
+
+            while (checks.size() > 0&&(System.currentTimeMillis()-beginTime<time)) {
+                if (check) {
+                    //根据通知去判断结果
+                    for (Map.Entry<String, Boolean> entry : checkContent.entrySet()) {//循环遍历信号
+                        String notice = entry.getKey();
+                        URL url = allService.get(notice);
+                        if (url != null && !entry.getValue()) {//没检查，进行检查
+                            entry.setValue(true);//查找完后不再查找
+                            if (StringUtil.isNotEmpty(url.getAddress()) && StringUtil.isNotEmpty(name)) {
+                                List<String> check = null;
+                                check = checkService(url, checks, detail, str);//检查服务去
+                                checks.remove(url);//检查完后移除
+                                logger.info("remove------------------- url!!!!!"+url.getServiceInterface());
+
+                            }
+                        }
+                    }
+                }
+                Thread.sleep(500);//休息500毫秒
+            }
+            logger.info("checks size!!!!!"+checks.size());
+            return detail;
+        }
+    }
 }

matrix-gracefulshutdown/src/main/java/com/secoo/mall/dubbo/spring/boot/autoconfigure/MatrixGracefulShutDownAutoConfiguration.java

 package com.secoo.mall.dubbo.spring.boot.autoconfigure;
 
 import com.secoo.mall.common.core.service.UpDatas;
+import com.secoo.mall.common.util.string.StringUtil;
 import com.secoo.mall.dubbo.monitor.config.ConfigCenter;
 import com.secoo.mall.dubbo.monitor.dubbo.service.ProviderService;
 import com.secoo.mall.dubbo.monitor.dubbo.service.RegistryServerSync;
@@ -39,21 +40,29 @@ public class MatrixGracefulShutDownAutoConfiguration {
         DubboUpdata  transport=  new DubboUpdata();
         URL url =configCenter.formUrl(configCenter.registryAddress, configCenter.registryGroup, configCenter.username, configCenter.password);
         transport.setUrl(url);
+        if(StringUtil.isEmpty(configCenter.patch)){
+            configCenter.patch="/monitorZ";
+        }
         transport.setPath(configCenter.patch);
         return transport;
 
     }
 
     @Bean
+    @ConditionalOnClass({ProviderService.class,RegistryServerSync.class})
     @ConditionalOnExpression("!T(org.springframework.util.StringUtils).isEmpty('${dubbo.registry.address:}')")
-    DubboCustomerDownHock createDubboConsumerDownHock(ProviderService providerService) {
-        return new DubboCustomerDownHock(providerService);
+    DubboCustomerDownHock createDubboConsumerDownHock(ProviderService providerService, RegistryServerSync registryServerSync) {
+        DubboCustomerDownHock hock= new DubboCustomerDownHock(providerService);
+        registryServerSync.addObserver(hock);
+        hock.setRegistryServerSync(registryServerSync);
+        return  hock;
     }
 
     @Bean
     @ConditionalOnExpression("!T(org.springframework.util.StringUtils).isEmpty('${dubbo.registry.address:}')")
     public RegistryServerSync createSynObject(Registry registry) {
-        return new RegistryServerSync(registry);
+        RegistryServerSync r=  new RegistryServerSync(registry);
+        return r;
     }