服务发现任务图

SwitchRefresher故障转移刷新

查看缓存目录下的/failover/00-00---000-VIPSRV_FAILOVER_SWITCH-000---00-00文件，看是否有改变，有改变就要设置相应的属性来表示是否开启故障转移模式，也就是做容灾备份的。

           @Override
            public void run() {
                try {//是否有转移文件
                    File switchFile = new File(failoverDir + UtilAndComs.FAILOVER_SWITCH);
                    if (!switchFile.exists()) {
                        switchParams.put("failover-mode", "false");
                        NAMING_LOGGER.debug("failover switch is not found, " + switchFile.getName());
                        return;
                    }
                    //上次膝盖时间
                    long modified = switchFile.lastModified();
                    //有改变
                    if (lastModifiedMillis < modified) {
                        lastModifiedMillis = modified;
                        String failover = ConcurrentDiskUtil.getFileContent(failoverDir + UtilAndComs.FAILOVER_SWITCH,
                            Charset.defaultCharset().toString());//获取文件内容
                        if (!StringUtils.isEmpty(failover)) {
                            List<String> lines = Arrays.asList(failover.split(DiskCache.getLineSeparator()));
                            //根据内容进行容灾模式设置
                            for (String line : lines) {
                                String line1 = line.trim();
                                if ("1".equals(line1)) {
                                    switchParams.put("failover-mode", "true");
                                    NAMING_LOGGER.info("failover-mode is on");
                                    new FailoverFileReader().run();
                                } else if ("0".equals(line1)) {
                                    switchParams.put("failover-mode", "false");
                                    NAMING_LOGGER.info("failover-mode is off");
                                }
                            }
                        } else {
                            switchParams.put("failover-mode", "false");
                        }
                    }
    
                } catch (Throwable e) {
                    NAMING_LOGGER.error("[NA] failed to read failover switch.", e);
                }
            }

FailoverReactor初始化备份

FailoverReactor中初始化会延迟10秒后做一次备份，如果发现缓存的/failover目录下有文件，就会写入本地。

    executorService.schedule(new Runnable() {
                @Override
                public void run() {
                    try {
                        File cacheDir = new File(failoverDir);
    
                        if (!cacheDir.exists() && !cacheDir.mkdirs()) {
                            throw new IllegalStateException("failed to create cache dir: " + failoverDir);
                        }
    
                        File[] files = cacheDir.listFiles();
                        if (files == null || files.length <= 0) {
                            new DiskFileWriter().run();
                        }
                    } catch (Throwable e) {
                        NAMING_LOGGER.error("[NA] failed to backup file on startup.", e);
                    }
    
                }
            }, 10000L, TimeUnit.MILLISECONDS);

DiskFileWriter

默认是一天写一次，就是把服务信息写入本地。

    class DiskFileWriter extends TimerTask {
            @Override
            public void run() {
                Map<String, ServiceInfo> map = hostReactor.getServiceInfoMap();
                for (Map.Entry<String, ServiceInfo> entry : map.entrySet()) {
                    ServiceInfo serviceInfo = entry.getValue();
                    if (StringUtils.equals(serviceInfo.getKey(), UtilAndComs.ALL_IPS) || StringUtils.equals(
                        serviceInfo.getName(), UtilAndComs.ENV_LIST_KEY)
                        || StringUtils.equals(serviceInfo.getName(), "00-00---000-ENV_CONFIGS-000---00-00")
                        || StringUtils.equals(serviceInfo.getName(), "vipclient.properties")
                        || StringUtils.equals(serviceInfo.getName(), "00-00---000-ALL_HOSTS-000---00-00")) {
                        continue;
                    }
    
                    DiskCache.write(serviceInfo, failoverDir);
                }
            }
        }

PushReceiver

接受服务端发来的UDP报文，然后根据pushPacket.type要求作出响应，在服务更新的时候会带上UDP端口号，这样就可以接受啦。

     @Override
        public void run() {
            while (true) {
                try {
                    // byte[] is initialized with 0 full filled by default
                    byte[] buffer = new byte[UDP_MSS];
                    DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
    
                    udpSocket.receive(packet);
    
                    String json = new String(IoUtils.tryDecompress(packet.getData()), "UTF-8").trim();
                    NAMING_LOGGER.info("received push data: " + json + " from " + packet.getAddress().toString());
    
                    PushPacket pushPacket = JSON.parseObject(json, PushPacket.class);
                    String ack;
                    if ("dom".equals(pushPacket.type) || "service".equals(pushPacket.type)) {
                        hostReactor.processServiceJSON(pushPacket.data);
    
                        // send ack to server
                        ack = "{\"type\": \"push-ack\""
                            + ", \"lastRefTime\":\"" + pushPacket.lastRefTime
                            + "\", \"data\":" + "\"\"}";
                    } else if ("dump".equals(pushPacket.type)) {
                        // dump data to server
                        ack = "{\"type\": \"dump-ack\""
                            + ", \"lastRefTime\": \"" + pushPacket.lastRefTime
                            + "\", \"data\":" + "\""
                            + StringUtils.escapeJavaScript(JSON.toJSONString(hostReactor.getServiceInfoMap()))
                            + "\"}";
                    } else {
                        // do nothing send ack only
                        ack = "{\"type\": \"unknown-ack\""
                            + ", \"lastRefTime\":\"" + pushPacket.lastRefTime
                            + "\", \"data\":" + "\"\"}";
                    }
    
                    udpSocket.send(new DatagramPacket(ack.getBytes(Charset.forName("UTF-8")),
                        ack.getBytes(Charset.forName("UTF-8")).length, packet.getSocketAddress()));
                } catch (Exception e) {
                    NAMING_LOGGER.error("[NA] error while receiving push data", e);
                }
            }
        }

好了，今天就到这里了，希望对学习理解有帮助，大神看见勿喷，仅为自己的学习理解，能力有限，请多包涵。