Fork me on GitHub

运维/shell学习

Shell 概念

图形界面和命令行只是架设在用户和内核之间的一座桥梁。

由于安全、复杂、繁琐等原因，用户不能直接接触内核（也没有必要），需要另外再开发一个程序，让用户直接使用这个程序；该程序的作用就是接收用户的操作（点击图标、输入命令），并进行简单的处理，然后再传递给内核，这样用户就能间接地使用操作系统内核了。你看，在用户和内核之间增加一层“代理”，既能简化用户的操作，又能保障内核的安全，何乐不为呢？

用户界面和命令行就是这个另外开发的程序，就是这层“代理”。在Linux下，这个命令行程序叫做 Shell。

简单来说： Shell 是一个应用程序，它连接了用户和 Linux 内核，让用户能够更加高效、安全、低成本地使用 Linux 内核，这就是 Shell 的本质。

Shell 是如何连接用户和内核的？

Shell 能够接收用户输入的命令，并对命令进行处理，处理完毕后再将结果反馈给用户，比如输出到显示器、写入到文件等，这就是大部分读者对 Shell 的认知。你看，我一直都在使用 Shell，哪有使用内核哦？我也没有看到 Shell 将我和内核连接起来呀？！

其实，Shell 程序本身的功能是很弱的，比如文件操作、输入输出、进程管理等都得依赖内核。我们运行一个命令，大部分情况下 Shell 都会去调用内核暴露出来的接口，这就是在使用内核，只是这个过程被 Shell 隐藏了起来，它自己在背后默默进行，我们看不到而已。

接口其实就是一个一个的函数，使用内核就是调用这些函数。这就是使用内核的全部内容了吗？嗯，是的！除了函数，你没有别的途径使用内核。

比如，我们都知道在 Shell 中输入cat log.txt命令就可以查看 log.txt 文件中的内容，然而，log.txt 放在磁盘的哪个位置？分成了几个数据块？在哪里开始？在哪里终止？如何操作探头读取它？这些底层细节 Shell 统统不知道的，它只能去调用内核提供的 open() 和 read() 函数，告诉内核我要读取 log.txt 文件，请帮助我，然后内核就乖乖地按照 Shell 的吩咐去读取文件了，并将读取到的文件内容交给 Shell，最后再由 Shell 呈现给用户（其实呈现到显示器上还得依赖内核）。整个过程中 Shell 就是一个“中间商”，它在用户和内核之间“倒卖”数据，只是用户不知道罢了。

Shell 还能连接其它程序

在 Shell 中输入的命令，有一部分是 Shell 本身自带的，这叫做内置命令；有一部分是其它的应用程序（一个程序就是一个命令），这叫做外部命令。

Shell 本身支持的命令并不多，功能也有限，但是 Shell 可以调用其他的程序，每个程序就是一个命令，这使得 Shell 命令的数量可以无限扩展，其结果就是 Shell 的功能非常强大，完全能够胜任 Linux 的日常管理工作，如文本或字符串检索、文件的查找或创建、大规模软件的自动部署、更改系统设置、监控服务器性能、发送报警邮件、抓取网页内容、压缩文件等。

更加惊讶的是，Shell 还可以让多个外部程序发生连接，在它们之间很方便地传递数据，也就是把一个程序的输出结果传递给另一个程序作为输入。

大家所说的 Shell 强大，并不是 Shell 本身功能丰富，而是它擅长使用和组织其他的程序。Shell 就是一个领导者，这正是 Shell 的魅力所在。

可以将 Shell 在整个 Linux 系统中的地位描述成下图所示的样子。注意“用户”和“其它应用程序”是通过虚线连接的，因为用户启动 Linux 后直接面对的是 Shell，通过 Shell 才能运行其它的应用程序。

Shell在整个Linux系统中的地位示意图

Shell 也支持编程

Shell 是一种脚本语言

任何代码最终都要被“翻译”成二进制的形式才能在计算机中执行。

有的编程语言，如 C/C++、Pascal、Go语言、汇编等，必须在程序运行之前将所有代码都翻译成二进制形式，也就是生成可执行文件，用户拿到的是最终生成的可执行文件，看不到源码。

这个过程叫做编译（Compile），这样的编程语言叫做编译型语言，完成编译过程的软件叫做编译器（Compiler）。

而有的编程语言，如 Shell、JavaScript、Python、PHP等，需要一边执行一边翻译，不会生成任何可执行文件，用户必须拿到源码才能运行程序。程序运行后会即时翻译，翻译完一部分执行一部分，不用等到所有代码都翻译完。

这个过程叫做解释，这样的编程语言叫做解释型语言或者脚本语言（Script），完成解释过程的软件叫做解释器。

编译型语言的优点是执行速度快、对硬件要求低、保密性好，适合开发操作系统、大型应用程序、数据库等。

脚本语言的优点是使用灵活、部署容易、跨平台性好，非常适合 Web 开发以及小工具的制作。

Shell 就是一种脚本语言，我们编写完源码后不用编译，直接运行源码即可。

shell使用

Shell 命令的基本格式如下：

command [选项] [参数]

短格式选项和长格式选项

Linux 的选项又分为短格式选项和长格式选项。

  • 短格式选项是长格式选项的简写，用一个减号-和一个字母表示，例如ls -l。

  • 长格式选项是完整的英文单词，用两个减号--和一个单词表示，例如ls --all。

    一般情况下，短格式选项是长格式选项的缩写，也就是一个短格式选项会有对应的长格式选项。当然也有例外，比如 ls 命令的短格式选项-l就没有对应的长格式选项，所以具体的命令选项还需要通过帮助手册来查询。

使用参数

参数是命令的操作对象，一般情况下，文件、目录、用户和进程等都可以作为参数被命令操作。例如：

1
2
[mozhiyan@localhost demo]$ ls -l main.c
-rw-rw-r--. 1 mozhiyan mozhiyan 650 4月  10 11:06 main.c

但是为什么一开始 ls 命令可以省略参数？那是因为有默认参数。命令一般都需要加入参数，用于指定命令操作的对象是谁。

Linux Shell命令提示符

不同的 Linux 发行版使用的提示符格式大同小异，例如在 CentOS 中，默认的提示符类似下面这样：

[mozhiyan@localhost ~]$

各个部分的含义如下：

  • []是提示符的分隔符号，没有特殊含义。
  • mozhiyan表示当前登录的用户，我现在使用的是 mozhiyan 用户登录。
  • @是分隔符号，没有特殊含义。
  • localhost表示当前系统的简写主机名（完整主机名是 localhost.localdomain）。
  • ~代表用户当前所在的目录为主目录（home 目录）。如果用户当前位于主目录下的 bin 目录中，那么这里显示的就是bin。
  • $是命令提示符。Linux 用这个符号标识登录的用户权限等级：如果是超级用户（root 用户），提示符就是#；如果是普通用户，提示符就是$。

第二层命令提示符

有些命令不能在一行内输入完成，需要换行，这个时候就会看到第二层命令提示符。第二层命令提示符默认为>

1
2
3
4
5
6
7
8
9
[mozhiyan@localhost ~]$ echo "Shell教程"
Shell教程
[mozhiyan@localhost ~]$ echo "
> http://
> c.biancheng.net
> "

http://
c.biancheng.net

第一个 echo 命令在一行内输入完成，不会出现第二层提示符。第二个 echo 命令需要多行才能输入完成，提示符>
echo 命令用来输出一个字符串。字符串是一组由包围起来的字符序列，echo 将第一个作为字符串的开端，将第二个作为字符串的结尾。对于第二个 echo 命令，我们将字符串分成多行，echo 遇到第一个认为是不完整的字符串，所以会继续等待用户输入，直到遇见第二个。

简单编程

1
2
3
4
5
6
7
8
9
10
11
12
13
几乎所有编程语言的教程都是从使用著名的“Hello World”开始的，出于对这种传统的尊重（或者说落入俗套），我们的第一个 Shell 脚本也输出“Hello World”。

打开文本编辑器，新建一个文本文件，并命名为 test.sh。
扩展名sh代表 shell，扩展名并不影响脚本执行，见名知意就好，如果你用 php 写 shell 脚本，扩展名就用php好了。

在 test.sh 中输入代码：
#!/bin/bash
echo "Hello World !"  #这是一条语句
第 1 行的#!是一个约定的标记，它告诉系统这个脚本需要什么解释器来执行，即使用哪一种 Shell；后面的/bin/bash就是指明了解释器的具体位置。

第 2 行的 echo 命令用于向标准输出文件（Standard Output，stdout，一般就是指显示器）输出文本。在.sh文件中使用命令与在终端直接输入命令的效果是一样的。

第 2 行的#及其后面的内容是注释。Shell 脚本中所有以#开头的都是注释（当然以#!开头的除外）。写脚本的时候，多写注释是非常有必要的，以方便其他人能看懂你的脚本，也方便后期自己维护时看懂自己的脚本——实际上，即便是自己写的脚本，在经过一段时间后也很容易忘记。

下面给出了一段稍微复杂的 Shell 脚本：

1
2
3
4
5
#!/bin/bash
#Copyright (c) http://c.biancheng.net/shell/
echo "What is your name?"
read PERSON
echo "Hello, $PERSON"

第 5 行中表示从终端读取用户输入的数据，并赋值给 PERSON 变量。read 命令用来从标准输入文件（Standard Input，stdin，一般就是指键盘）读取用户输入的数据。
第 6 行表示输出变量 PERSON 的内容。注意在变量名前边要加上$，否则变量名会作为字符串的一部分处理。

在新进程中运行 Shell 脚本

在新进程中运行 Shell 脚本有多种方法。

1) 将 Shell 脚本作为程序运行

Shell 脚本也是一种解释执行的程序，可以在终端直接调用（需要使用 chmod 命令给 Shell 脚本加上执行权限），如下所示：

1
2
3
4
[mozhiyan@localhost ~]$ cd demo                #切换到 test.sh 所在的目录
[mozhiyan@localhost demo]$ chmod +x ./test.sh  #给脚本添加执行权限
[mozhiyan@localhost demo]$ ./test.sh           #执行脚本文件
Hello World !                                  #运行结果

第 2 行中，chmod +x表示给 test.sh 增加执行权限。

第 3 行中，./表示当前目录，整条命令的意思是执行当前目录下的 test.sh 脚本。如果不写./，Linux 会到系统路径（由 PATH 环境变量指定）下查找 test.sh，而系统路径下显然不存在这个脚本，所以会执行失败。

通过这种方式运行脚本，脚本文件第一行的#!/bin/bash一定要写对，好让系统查找到正确的解释器。

在当前进程中运行 Shell 脚本

这里需要引入一个新的命令——source 命令。source 是 Shell 内置命令 的一种，它会读取脚本文件中的代码，并依次执行所有语句。你也可以理解为，source 命令会强制执行脚本文件中的全部命令，而忽略脚本文件的权限。

source 命令的用法为：

source filename

也可以简写为：

. filename

两种写法的效果相同。对于第二种写法，注意点号.

展开全文 >>

framework/SpringCloud/Nacos/Nacos名词

Nacos名词

NamingFactory

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

Naming ： 提供 通过命名 来查找服务
那么这个NamingFactory就是创建Naming的
    
 /**
     * Create a new naming service.
     *
     * @param serverList server list
     * @return new naming service
     * @throws NacosException nacos exception
     */
    public static NamingService createNamingService(String serverList) throws NacosException {
        try {
            Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.naming.NacosNamingService");
            Constructor constructor = driverImplClass.getConstructor(String.class);
            NamingService vendorImpl = (NamingService) constructor.newInstance(serverList);
            return vendorImpl;
        } catch (Throwable e) {
            throw new NacosException(NacosException.CLIENT_INVALID_PARAM, e);
        }
    }
    
    /**
     * Create a new naming service.
     *
     * @param properties naming service properties
     * @return new naming service
     * @throws NacosException nacos exception
     */
    public static NamingService createNamingService(Properties properties) throws NacosException {
        try {
            Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.naming.NacosNamingService");
            Constructor constructor = driverImplClass.getConstructor(Properties.class);
            NamingService vendorImpl = (NamingService) constructor.newInstance(properties);
            return vendorImpl;
        } catch (Throwable e) {
            throw new NacosException(NacosException.CLIENT_INVALID_PARAM, e);
        }
    }

NacosNamingService

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
继承  

/**
     * 每个命名服务（注册中心）应该有不同的命名空间。
     */
    private String namespace;
    
    private String endpoint;
    
    private String serverList;
    
    private String cacheDir;
    
    private String logName;
    
    private HostReactor hostReactor;
    
    private BeatReactor beatReactor;
    
    private EventDispatcher eventDispatcher;
    
    private NamingProxy serverProxy;
    

private void init(Properties properties) throws NacosException {
        ValidatorUtils.checkInitParam(properties);
        this.namespace = InitUtils.initNamespaceForNaming(properties);
        InitUtils.initSerialization();
        initServerAddr(properties);
        InitUtils.initWebRootContext();
        initCacheDir();
        initLogName(properties);
        
        this.eventDispatcher = new EventDispatcher();
        this.serverProxy = new NamingProxy(this.namespace, this.endpoint, this.serverList, properties);
        this.beatReactor = new BeatReactor(this.serverProxy, initClientBeatThreadCount(properties));
        this.hostReactor = new HostReactor(this.eventDispatcher, this.serverProxy, beatReactor, this.cacheDir,
                isLoadCacheAtStart(properties), initPollingThreadCount(properties));
    }

initNamespaceForNaming

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

public static String initNamespaceForNaming(Properties properties) {
       String tmpNamespace = null;
       
       String isUseCloudNamespaceParsing = properties.getProperty(PropertyKeyConst.IS_USE_CLOUD_NAMESPACE_PARSING,
               System.getProperty(SystemPropertyKeyConst.IS_USE_CLOUD_NAMESPACE_PARSING,
                       String.valueOf(Constants.DEFAULT_USE_CLOUD_NAMESPACE_PARSING)));
       
       if (Boolean.parseBoolean(isUseCloudNamespaceParsing)) {
           
           tmpNamespace = TenantUtil.getUserTenantForAns();
           //如果tmpNamespace=null 则从call里面获取
           tmpNamespace = TemplateUtils.stringEmptyAndThenExecute(tmpNamespace, new Callable<String>() {
               @Override
               public String call() {
                   String namespace = System.getProperty(SystemPropertyKeyConst.ANS_NAMESPACE);
                   LogUtils.NAMING_LOGGER.info("initializer namespace from System Property :" + namespace);
                   return namespace;
               }
           });
           
           tmpNamespace = TemplateUtils.stringEmptyAndThenExecute(tmpNamespace, new Callable<String>() {
               @Override
               public String call() {
                   String namespace = System.getenv(PropertyKeyConst.SystemEnv.ALIBABA_ALIWARE_NAMESPACE);
                   LogUtils.NAMING_LOGGER.info("initializer namespace from System Environment :" + namespace);
                   return namespace;
               }
           });
       }
       
       tmpNamespace = TemplateUtils.stringEmptyAndThenExecute(tmpNamespace, new Callable<String>() {
           @Override
           public String call() {
               String namespace = System.getProperty(PropertyKeyConst.NAMESPACE);
               LogUtils.NAMING_LOGGER.info("initializer namespace from System Property :" + namespace);
               return namespace;
           }
       });
       
       if (StringUtils.isEmpty(tmpNamespace) && properties != null) {
           tmpNamespace = properties.getProperty(PropertyKeyConst.NAMESPACE);
       }
       
    
    //最后   public static final String DEFAULT_NAMESPACE_ID = "public";默认是public
       tmpNamespace = TemplateUtils.stringEmptyAndThenExecute(tmpNamespace, new Callable<String>() {
           @Override
           public String call() {
               return UtilAndComs.DEFAULT_NAMESPACE_ID;
           }
       });
       return tmpNamespace;
   }

initSerialization

1
2
3
4
5
6
7
8
9
10

* <p>
   * 子类实现类中的静态代码串中已经向Jackson进行了注册，但是由于classloader的原因，只有当 该子类被使用的时候，才会加载该类。这可能会导致Jackson先进性反序列化，再注册子类，从而导致 反序列化失败。
   * </p>
   */
  public static void initSerialization() {
      // TODO register in implementation class or remove subType
      JacksonUtils.registerSubtype(NoneSelector.class, SelectorType.none.name());
      JacksonUtils.registerSubtype(ExpressionSelector.class, SelectorType.label.name());
  }

initServerAddr

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

private void initServerAddr(Properties properties) {
       serverList = properties.getProperty(PropertyKeyConst.SERVER_ADDR);
       endpoint = InitUtils.initEndpoint(properties);
       if (StringUtils.isNotEmpty(endpoint)) {
           serverList = "";
       }
   }

 public static String initEndpoint(final Properties properties) {
       if (properties == null) {
           
           return "";
       }
       // Whether to enable domain name resolution rules
       String isUseEndpointRuleParsing = properties.getProperty(PropertyKeyConst.IS_USE_ENDPOINT_PARSING_RULE,
               System.getProperty(SystemPropertyKeyConst.IS_USE_ENDPOINT_PARSING_RULE,
                       String.valueOf(ParamUtil.USE_ENDPOINT_PARSING_RULE_DEFAULT_VALUE)));
       
       boolean isUseEndpointParsingRule = Boolean.parseBoolean(isUseEndpointRuleParsing);
       String endpointUrl;
       if (isUseEndpointParsingRule) {
           // Get the set domain name information
           endpointUrl = ParamUtil.parsingEndpointRule(properties.getProperty(PropertyKeyConst.ENDPOINT));
           if (StringUtils.isBlank(endpointUrl)) {
               return "";
           }
       } else {
           endpointUrl = properties.getProperty(PropertyKeyConst.ENDPOINT);
       }
       
       if (StringUtils.isBlank(endpointUrl)) {
           return "";
       }
       
       String endpointPort = TemplateUtils
               .stringEmptyAndThenExecute(System.getenv(PropertyKeyConst.SystemEnv.ALIBABA_ALIWARE_ENDPOINT_PORT),
                       new Callable<String>() {
                           @Override
                           public String call() {
                               
                               return properties.getProperty(PropertyKeyConst.ENDPOINT_PORT);
                           }
                       });
       
       endpointPort = TemplateUtils.stringEmptyAndThenExecute(endpointPort, new Callable<String>() {
           @Override
           public String call() {
               return "8080";
           }
       });
       
       return endpointUrl + ":" + endpointPort;
   }

initWebRootContext

1
2
3
4
5
6
7
8
9
10
11
12
13
14

public static void initWebRootContext() {
    // support the web context with ali-yun if the app deploy by EDAS
    final String webContext = System.getProperty(SystemPropertyKeyConst.NAMING_WEB_CONTEXT);
    TemplateUtils.stringNotEmptyAndThenExecute(webContext, new Runnable() {
        @Override
        public void run() {
            UtilAndComs.webContext = webContext.indexOf("/") > -1 ? webContext : "/" + webContext;
            
            UtilAndComs.nacosUrlBase = UtilAndComs.webContext + "/v1/ns";
            UtilAndComs.nacosUrlInstance = UtilAndComs.nacosUrlBase + "/instance";
        }
    });
}

initCacheDir

1
2
3
4
5
6
7

private void initCacheDir() {
        cacheDir = System.getProperty("com.alibaba.nacos.naming.cache.dir");
        if (StringUtils.isEmpty(cacheDir)) {
            cacheDir = System.getProperty("user.home") + "/nacos/naming/" + namespace;
        }
    }

initLogName

1
2
3
4
5
6
7
8
9
10
11
12
13

private void initLogName(Properties properties) {
       logName = System.getProperty(UtilAndComs.NACOS_NAMING_LOG_NAME);
       if (StringUtils.isEmpty(logName)) {
           
           if (properties != null && StringUtils
                   .isNotEmpty(properties.getProperty(UtilAndComs.NACOS_NAMING_LOG_NAME))) {
               logName = properties.getProperty(UtilAndComs.NACOS_NAMING_LOG_NAME);
           } else {
               logName = "naming.log";
           }
       }
   }

EventDispatcher

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
// 事件分发器,那几可以分发，可以注册事件，此分发器 之分发 服务信息 的事件
EventDispatcher
 private ExecutorService executor = null;
    
    private final BlockingQueue<ServiceInfo> changedServices = new LinkedBlockingQueue<ServiceInfo>();
    
    private final ConcurrentMap<String, List<EventListener>> observerMap = new ConcurrentHashMap<String, List<EventListener>>();
    
    private volatile boolean closed = false;

public EventDispatcher() {
        
        //单线程
        this.executor = Executors.newSingleThreadExecutor(new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "com.alibaba.nacos.naming.client.listener");
                thread.setDaemon(true);
                
                return thread;
            }
        });
        
        this.executor.execute(new Notifier());
    }


 private class Notifier implements Runnable {
        
        @Override
        public void run() {
            
            // 循环调度
            while (!closed) {
                
                ServiceInfo serviceInfo = null;
                try {
                    //弹出一个  （服务信息） 
                    serviceInfo = changedServices.poll(5, TimeUnit.MINUTES);
                } catch (Exception ignore) {
                }
                
                if (serviceInfo == null) {
                    continue;
                }
                
                try {
                    List<EventListener> listeners = observerMap.get(serviceInfo.getKey());
                    
                    if (!CollectionUtils.isEmpty(listeners)) {
                        for (EventListener listener : listeners) {
                            // 这是一个不能修改的集合，已修改就报错
                            List<Instance> hosts = Collections.unmodifiableList(serviceInfo.getHosts());
                            
                            listener.onEvent(new NamingEvent(serviceInfo.getName(), serviceInfo.getGroupName(),
                                    serviceInfo.getClusters(), hosts));
                        }
                    }
                    
                } catch (Exception e) {
                    NAMING_LOGGER.error("[NA] notify error for service: " + serviceInfo.getName() + ", clusters: "
                            + serviceInfo.getClusters(), e);
                }
            }
        }
    }

NamingProxy

注册中心的代理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
private final NacosRestTemplate nacosRestTemplate = NamingHttpClientManager.getInstance().getNacosRestTemplate();
   
   private static final int DEFAULT_SERVER_PORT = 8848;
   
   private int serverPort = DEFAULT_SERVER_PORT;
   
   private final String namespaceId;
   
   private final String endpoint;
   
   private String nacosDomain;
   
   private List<String> serverList;
   
   private List<String> serversFromEndpoint = new ArrayList<String>();
   
   private final SecurityProxy securityProxy;
   
   private long lastSrvRefTime = 0L;
   
   private final long vipSrvRefInterMillis = TimeUnit.SECONDS.toMillis(30);
   
   private final long securityInfoRefreshIntervalMills = TimeUnit.SECONDS.toMillis(5);
   
   private Properties properties;
   
   private ScheduledExecutorService executorService;
   


public NamingProxy(String namespaceId, String endpoint, String serverList, Properties properties) {
       
       this.securityProxy = new SecurityProxy(properties, nacosRestTemplate);
       this.properties = properties;
       this.setServerPort(DEFAULT_SERVER_PORT);
       this.namespaceId = namespaceId;
       this.endpoint = endpoint;
       if (StringUtils.isNotEmpty(serverList)) {
           this.serverList = Arrays.asList(serverList.split(","));
           if (this.serverList.size() == 1) {
               this.nacosDomain = serverList;
           }
       }
       this.initRefreshTask();
   }

initRefreshTask

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

private void initRefreshTask() {
       
       this.executorService = new ScheduledThreadPoolExecutor(2, new ThreadFactory() {
           @Override
           public Thread newThread(Runnable r) {
               Thread t = new Thread(r);
               t.setName("com.alibaba.nacos.client.naming.updater");
               t.setDaemon(true);
               return t;
           }
       });
       
       this.executorService.scheduleWithFixedDelay(new Runnable() {
           @Override
           public void run() {
               refreshSrvIfNeed();
           }
       }, 0, vipSrvRefInterMillis, TimeUnit.MILLISECONDS);
       
       this.executorService.scheduleWithFixedDelay(new Runnable() {
           @Override
           public void run() {
               securityProxy.login(getServerList());
           }
       }, 0, securityInfoRefreshIntervalMills, TimeUnit.MILLISECONDS);
       
       refreshSrvIfNeed();
       this.securityProxy.login(getServerList());
   }



private void refreshSrvIfNeed() {
       try {
           
           //serverList==null 就返回
           if (!CollectionUtils.isEmpty(serverList)) {
               NAMING_LOGGER.debug("server list provided by user: " + serverList);
               return;
           }
           
           if (System.currentTimeMillis() - lastSrvRefTime < vipSrvRefInterMillis) {
               
               //当前刷新时间间隔 少于 vipSrvRefInterMillis
               return;
           }
           
           List<String> list = getServerListFromEndpoint();
           
           if (CollectionUtils.isEmpty(list)) {
               throw new Exception("Can not acquire Nacos list");
           }
           
           if (!CollectionUtils.isEqualCollection(list, serversFromEndpoint)) {
               NAMING_LOGGER.info("[SERVER-LIST] server list is updated: " + list);
           }
           
           serversFromEndpoint = list;
           lastSrvRefTime = System.currentTimeMillis();
       } catch (Throwable e) {
           NAMING_LOGGER.warn("failed to update server list", e);
       }
   }


public boolean login(List<String> servers) {
       
       try {
           //TTL是 Time To Live的缩写，该字段指定IP包被路由器丢弃之前允许通过的最大网段数量。TTL是IPv4包头的一个8 bit字段。
           // tokenTtl：token的生存时间值，tokenRefreshWindow 刷新安全信息的时间窗口（秒）。
           if ((System.currentTimeMillis() - lastRefreshTime) < TimeUnit.SECONDS
                   .toMillis(tokenTtl - tokenRefreshWindow)) {
               //防止反复刷新的？
               return true;
           }
           
           for (String server : servers) {
               if (login(server)) {
                   lastRefreshTime = System.currentTimeMillis();
                   return true;
               }
           }
       } catch (Throwable ignore) {
       }
       
       return false;
   }
getServerListFromEndpoint
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

public List<String> getServerListFromEndpoint() {
       
       try {
           String urlString = "http://" + endpoint + "/nacos/serverlist";
           Header header = builderHeader();
           HttpRestResult<String> restResult = nacosRestTemplate.get(urlString, header, Query.EMPTY, String.class);
           if (!restResult.ok()) {
               throw new IOException(
                       "Error while requesting: " + urlString + "'. Server returned: " + restResult.getCode());
           }
           
           String content = restResult.getData();
           List<String> list = new ArrayList<String>();
           for (String line : IoUtils.readLines(new StringReader(content))) {
               if (!line.trim().isEmpty()) {
                   list.add(line.trim());
               }
           }
           
           return list;
           
       } catch (Exception e) {
           e.printStackTrace();
       }
       
       return null;
   }
login
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

public boolean login(String server) {
      
      if (StringUtils.isNotBlank(username)) {
          Map<String, String> params = new HashMap<String, String>(2);
          Map<String, String> bodyMap = new HashMap<String, String>(2);
          params.put("username", username);
          bodyMap.put("password", password);
          String url = "http://" + server + contextPath + LOGIN_URL;
          
          if (server.contains(Constants.HTTP_PREFIX)) {
              url = server + contextPath + LOGIN_URL;
          }
          try {
              HttpRestResult<String> restResult = nacosRestTemplate
                      .postForm(url, Header.EMPTY, params, bodyMap, String.class);
              if (!restResult.ok()) {
                  SECURITY_LOGGER.error("login failed: {}", JacksonUtils.toJson(restResult));
                  return false;
              }
              JsonNode obj = JacksonUtils.toObj(restResult.getData());
              if (obj.has(Constants.ACCESS_TOKEN)) {
                  accessToken = obj.get(Constants.ACCESS_TOKEN).asText();
                  tokenTtl = obj.get(Constants.TOKEN_TTL).asInt();
                  tokenRefreshWindow = tokenTtl / 10;
              }
          } catch (Exception e) {
              SECURITY_LOGGER.error("[SecurityProxy] login http request failed"
                      + " url: {}, params: {}, bodyMap: {}, errorMsg: {}", url, params, bodyMap, e.getMessage());
              return false;
          }
      }
      return true;
  }

SecurityProxy

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

SecurityProxy
    更新安全信息的安全代理。
    
   private static final Logger SECURITY_LOGGER = LoggerFactory.getLogger(SecurityProxy.class);
    
    private static final String LOGIN_URL = "/v1/auth/users/login";
    
    private final NacosRestTemplate nacosRestTemplate;
    
    private String contextPath;
    
    /**
     * User's name.
     */
    private final String username;
    
    /**
     * User's password.
     */
    private final String password;
    
    /**
     * A token to take with when sending request to Nacos server.
     */
    private String accessToken;
    
    /**
     * TTL of token in seconds.
     */
    private long tokenTtl;
    
    /**
     * Last timestamp refresh security info from server.
     */
    private long lastRefreshTime;
    
    /**
     * time window to refresh security info in seconds.
     */
    private long tokenRefreshWindow;   
    


    public SecurityProxy(Properties properties, NacosRestTemplate nacosRestTemplate) {
        username = properties.getProperty(PropertyKeyConst.USERNAME, StringUtils.EMPTY);
        password = properties.getProperty(PropertyKeyConst.PASSWORD, StringUtils.EMPTY);
        contextPath = properties.getProperty(PropertyKeyConst.CONTEXT_PATH, "/nacos");
        contextPath = contextPath.startsWith("/") ? contextPath : "/" + contextPath;
        this.nacosRestTemplate = nacosRestTemplate;
    }

BeatReactor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
心跳反应堆。


private final ScheduledExecutorService executorService;
    
    private final NamingProxy serverProxy;
    
    private boolean lightBeatEnabled = false;
    
    public final Map<String, BeatInfo> dom2Beat = new ConcurrentHashMap<String, BeatInfo>();

    public BeatReactor(NamingProxy serverProxy, int threadCount) {
        this.serverProxy = serverProxy;
        this.executorService = new ScheduledThreadPoolExecutor(threadCount, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("com.alibaba.nacos.naming.beat.sender");
                return thread;
            }
        });
    }

HostReactor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
//主反应器。

  private static final long DEFAULT_DELAY = 1000L;
    
    private static final long UPDATE_HOLD_INTERVAL = 5000L;
    
    private final Map<String, ScheduledFuture<?>> futureMap = new HashMap<String, ScheduledFuture<?>>();
    
    private final Map<String, ServiceInfo> serviceInfoMap;
    
    private final Map<String, Object> updatingMap;
    
    private final PushReceiver pushReceiver;
    
    private final EventDispatcher eventDispatcher;
    
    private final BeatReactor beatReactor;
    
    private final NamingProxy serverProxy;
    
    private final FailoverReactor failoverReactor;
    
    private final String cacheDir;
    
    private final ScheduledExecutorService executor;


public HostReactor(EventDispatcher eventDispatcher, NamingProxy serverProxy, BeatReactor beatReactor,
            String cacheDir, boolean loadCacheAtStart, int pollingThreadCount) {
        // init executorService
        this.executor = new ScheduledThreadPoolExecutor(pollingThreadCount, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("com.alibaba.nacos.client.naming.updater");
                return thread;
            }
        });
        this.eventDispatcher = eventDispatcher;
        this.beatReactor = beatReactor;
        this.serverProxy = serverProxy;
        this.cacheDir = cacheDir;
        if (loadCacheAtStart) {
            this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(DiskCache.read(this.cacheDir));
        } else {
            this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(16);
        }
        
        this.updatingMap = new ConcurrentHashMap<String, Object>();
        this.failoverReactor = new FailoverReactor(this, cacheDir);
        this.pushReceiver = new PushReceiver(this);
    }

PushReceiver

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
推动接收器。
  private static final Charset UTF_8 = Charset.forName("UTF-8");
    
    private static final int UDP_MSS = 64 * 1024;
    
    private ScheduledExecutorService executorService;
    
    private DatagramSocket udpSocket;
    
    private HostReactor hostReactor;
    
    private volatile boolean closed = false;
    
    public PushReceiver(HostReactor hostReactor) {
        try {
            this.hostReactor = hostReactor;
            this.udpSocket = new DatagramSocket();
            this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setDaemon(true);
                    thread.setName("com.alibaba.nacos.naming.push.receiver");
                    return thread;
                }
            });
            
            this.executorService.execute(this);
        } catch (Exception e) {
            NAMING_LOGGER.error("[NA] init udp socket failed", e);
        }
    }

//发布 服务信息修改 信息

   @Override
    public void run() {
        while (!closed) {
            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 = JacksonUtils.toObj(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(JacksonUtils.toJson(hostReactor.getServiceInfoMap()))
                            + "\"}";
                } else {
                    // do nothing send ack only
                    ack = "{\"type\": \"unknown-ack\"" + ", \"lastRefTime\":\"" + pushPacket.lastRefTime
                            + "\", \"data\":" + "\"\"}";
                }
                
                udpSocket.send(new DatagramPacket(ack.getBytes(UTF_8), ack.getBytes(UTF_8).length,
                        packet.getSocketAddress()));
            } catch (Exception e) {
                NAMING_LOGGER.error("[NA] error while receiving push data", e);
            }
        }
    }

FailoverReactor

故障转移反应器。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55


private final String failoverDir;
   
   private final HostReactor hostReactor;
   
   private final ScheduledExecutorService executorService;
   
   public FailoverReactor(HostReactor hostReactor, String cacheDir) {
       this.hostReactor = hostReactor;
       this.failoverDir = cacheDir + "/failover";
       // init executorService
       this.executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
           @Override
           public Thread newThread(Runnable r) {
               Thread thread = new Thread(r);
               thread.setDaemon(true);
               thread.setName("com.alibaba.nacos.naming.failover");
               return thread;
           }
       });
       this.init();
   }

public void init() {
       
    //这个好像是文件的
       executorService.scheduleWithFixedDelay(new SwitchRefresher(), 0L, 5000L, TimeUnit.MILLISECONDS);
       
    
    //写文件
       executorService.scheduleWithFixedDelay(new DiskFileWriter(), 30, DAY_PERIOD_MINUTES, TimeUnit.MINUTES);
       
       // backup file on startup if failover directory is empty.
       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);
   }

SwitchRefresher

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

class SwitchRefresher implements Runnable {
        
        long lastModifiedMillis = 0L;
        
        @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)) {
                        String[] lines = 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);
            }
        }
    }

DiskFileWriter

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

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);
            }
        }
    }

NamingService

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

这个接口更像一个注册中心，即提供注册，有提供获取，所以namingService的概念就有了，他就是一个注册中心

    
    //注册
void registerInstance(String serviceName, String ip, int port) throws NacosException;

   //从服务中注销实例。
  void deregisterInstance(String serviceName, String ip, int port) throws NacosException;

//获取指定服务名的所有实例
  List<Instance> getAllInstances(String serviceName) throws NacosException;

//选择健康的实例
List<Instance> selectInstances(String serviceName, boolean healthy) throws NacosException;


Instance selectOneHealthyInstance(String serviceName) throws NacosException
    
    
    //订阅，订阅服务接收实例更改事件。
  void subscribe(String serviceName, EventListener listener) throws NacosException;    


    void unsubscribe(String serviceName, EventListener listener) throws NacosException;

//从服务器获取所有服务名称。
ListView<String> getServicesOfServer(int pageNo, int pageSize) throws NacosException;

//获取当前客户端的所有订阅服务。
  List<ServiceInfo> getSubscribeServices() throws NacosException;

  /**
     * get server health status.
     *
     * @return is server healthy
     */
    String getServerStatus();
    
    /**
     * Shutdown the resource service.
     *
     * @throws NacosException exception.
     */
    void shutDown() throws NacosException;

服务端

ServiceManager

展开全文 >>

framework/SpringCloud/Nacos/Nacos简单使用

raft演示，如果集群一致性要求强，使用raft

http://thesecretlivesofdata.com/raft/

//一致性要求低，但是要高可用的话不就要选，浪费性能

Nacos简单使用

1
2
3
4
5
6
7
8
9
10
11
12

public class App {
    public static void main(String[] args) throws NacosException {
        Properties properties = new Properties();
        properties.setProperty("serverAddr", "21.34.53.5:8848,21.34.53.6:8848");
        properties.setProperty("namespace", "quickStart");
        NamingService naming = NamingFactory.createNamingService(properties);
        naming.registerInstance("nacos.test.3", "11.11.11.11", 8888, "TEST1");
        naming.registerInstance("nacos.test.3", "2.2.2.2", 9999, "DEFAULT");
        System.out.println(naming.getAllInstances("nacos.test.3"));
    }
}

api: 接口定义

client: 这个就是程序中引入的，用于访问nacos注册中心和配置中心的服务端

config: 配置中心服务端代码

naming：注册中心服务端代码

core：把config和naming部分公共代码抽取出来，config和naming都依赖于这个包

console: 这个依赖于naming和config，nacos有一个前端，就是和这个直接交互的

启动

单机版启动：找到nacos-console模块，然后再VM options这一栏

-Dnacos.standalone=true 设置为单机启动，单机启动不需要配置数据库，集群启动现在1.3.4必须要数据库支持

注册

客户端

NacosNamingService

1
2
3
4
5
6
/nacos/v1/ns/instance

    @Override
    public void registerInstance(String serviceName, String ip, int port, String clusterName) throws NacosException {
        registerInstance(serviceName, Constants.DEFAULT_GROUP, ip, port, clusterName);
    }
1
2
3
4
5
6
7
8
9
10
11
public void registerInstance(String serviceName, String groupName, String ip, int port, String clusterName)
           throws NacosException {
       
       Instance instance = new Instance();
       instance.setIp(ip);
       instance.setPort(port);
       instance.setWeight(1.0);
       instance.setClusterName(clusterName);
       
       registerInstance(serviceName, groupName, instance);
   }

registerInstance

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@Override
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
    String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
    if (instance.isEphemeral()) {
    //如果实例是短暂的。
        BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);
        beatReactor.addBeatInfo(groupedServiceName, beatInfo);
    }
    serverProxy.registerService(groupedServiceName, groupName, instance);
}



public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
    NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
    String key = buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());
    BeatInfo existBeat = null;
    //fix #1733
    if ((existBeat = dom2Beat.remove(key)) != null) {
        existBeat.setStopped(true);
    }
    dom2Beat.put(key, beatInfo);
    executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
    MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size());
}

BeatTask

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

class BeatTask implements Runnable {
       
       BeatInfo beatInfo;
       
       public BeatTask(BeatInfo beatInfo) {
           this.beatInfo = beatInfo;
       }
       
       @Override
       public void run() {
           if (beatInfo.isStopped()) {
               return;
           }
           long nextTime = beatInfo.getPeriod();
           try {
               JsonNode result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);
               long interval = result.get("clientBeatInterval").asLong();
               boolean lightBeatEnabled = false;
               if (result.has(CommonParams.LIGHT_BEAT_ENABLED)) {
                   lightBeatEnabled = result.get(CommonParams.LIGHT_BEAT_ENABLED).asBoolean();
               }
               BeatReactor.this.lightBeatEnabled = lightBeatEnabled;
               if (interval > 0) {
                   nextTime = interval;
               }
               int code = NamingResponseCode.OK;
               if (result.has(CommonParams.CODE)) {
                   code = result.get(CommonParams.CODE).asInt();
               }
               if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {
                   Instance instance = new Instance();
                   instance.setPort(beatInfo.getPort());
                   instance.setIp(beatInfo.getIp());
                   instance.setWeight(beatInfo.getWeight());
                   instance.setMetadata(beatInfo.getMetadata());
                   instance.setClusterName(beatInfo.getCluster());
                   instance.setServiceName(beatInfo.getServiceName());
                   instance.setInstanceId(instance.getInstanceId());
                   instance.setEphemeral(true);
                   try {
                       serverProxy.registerService(beatInfo.getServiceName(),
                               NamingUtils.getGroupName(beatInfo.getServiceName()), instance);
                   } catch (Exception ignore) {
                   }
               }
           } catch (NacosException ex) {
               NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, code: {}, msg: {}",
                       JacksonUtils.toJson(beatInfo), ex.getErrCode(), ex.getErrMsg());
               
           }
           executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);
       }
   }


public JsonNode sendBeat(BeatInfo beatInfo, boolean lightBeatEnabled) throws NacosException {
       
       if (NAMING_LOGGER.isDebugEnabled()) {
           NAMING_LOGGER.debug("[BEAT] {} sending beat to server: {}", namespaceId, beatInfo.toString());
       }
       Map<String, String> params = new HashMap<String, String>(8);
       Map<String, String> bodyMap = new HashMap<String, String>(2);
       if (!lightBeatEnabled) {
           bodyMap.put("beat", JacksonUtils.toJson(beatInfo));
       }
       params.put(CommonParams.NAMESPACE_ID, namespaceId);
       params.put(CommonParams.SERVICE_NAME, beatInfo.getServiceName());
       params.put(CommonParams.CLUSTER_NAME, beatInfo.getCluster());
       params.put("ip", beatInfo.getIp());
       params.put("port", String.valueOf(beatInfo.getPort()));
       String result = reqApi(UtilAndComs.nacosUrlBase + "/instance/beat", params, bodyMap, HttpMethod.PUT);
       return JacksonUtils.toObj(result);
   }


public String reqApi(String api, Map<String, String> params, Map<String, String> body, List<String> servers,
           String method) throws NacosException {
       
       params.put(CommonParams.NAMESPACE_ID, getNamespaceId());
       
       if (CollectionUtils.isEmpty(servers) && StringUtils.isEmpty(nacosDomain)) {
           throw new NacosException(NacosException.INVALID_PARAM, "no server available");
       }
       
       NacosException exception = new NacosException();
       
       if (servers != null && !servers.isEmpty()) {
           
           Random random = new Random(System.currentTimeMillis());
           int index = random.nextInt(servers.size());
           
           for (int i = 0; i < servers.size(); i++) {
               String server = servers.get(index);
               try {
                   return callServer(api, params, body, server, method);
               } catch (NacosException e) {
                   exception = e;
                   if (NAMING_LOGGER.isDebugEnabled()) {
                       NAMING_LOGGER.debug("request {} failed.", server, e);
                   }
               }
               index = (index + 1) % servers.size();
           }
       }
       
       if (StringUtils.isNotBlank(nacosDomain)) {
           for (int i = 0; i < UtilAndComs.REQUEST_DOMAIN_RETRY_COUNT; i++) {
               try {
                   return callServer(api, params, body, nacosDomain, method);
               } catch (NacosException e) {
                   exception = e;
                   if (NAMING_LOGGER.isDebugEnabled()) {
                       NAMING_LOGGER.debug("request {} failed.", nacosDomain, e);
                   }
               }
           }
       }
       
       NAMING_LOGGER.error("request: {} failed, servers: {}, code: {}, msg: {}", api, servers, exception.getErrCode(),
               exception.getErrMsg());
       
       throw new NacosException(exception.getErrCode(),
               "failed to req API:" + api + " after all servers(" + servers + ") tried: " + exception.getMessage());
       
   }


  public String callServer(String api, Map<String, String> params, Map<String, String> body, String curServer,
           String method) throws NacosException {
       long start = System.currentTimeMillis();
       long end = 0;
       injectSecurityInfo(params);
       Header header = builderHeader();
       
       String url;
       if (curServer.startsWith(UtilAndComs.HTTPS) || curServer.startsWith(UtilAndComs.HTTP)) {
           url = curServer + api;
       } else {
           if (!curServer.contains(UtilAndComs.SERVER_ADDR_IP_SPLITER)) {
               curServer = curServer + UtilAndComs.SERVER_ADDR_IP_SPLITER + serverPort;
           }
           url = NamingHttpClientManager.getInstance().getPrefix() + curServer + api;
       }
       
       try {
           HttpRestResult<String> restResult = nacosRestTemplate
                   .exchangeForm(url, header, params, body, method, String.class);
           end = System.currentTimeMillis();
           
           MetricsMonitor.getNamingRequestMonitor(method, url, String.valueOf(restResult.getCode()))
                   .observe(end - start);
           
           if (restResult.ok()) {
               return restResult.getData();
           }
           if (HttpStatus.SC_NOT_MODIFIED == restResult.getCode()) {
               return StringUtils.EMPTY;
           }
           throw new NacosException(restResult.getCode(), restResult.getMessage());
       } catch (Exception e) {
           NAMING_LOGGER.error("[NA] failed to request", e);
           throw new NacosException(NacosException.SERVER_ERROR, e);
       }
   }
registerService

BeatReactor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
       
       NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", namespaceId, serviceName,
               instance);
       
       final Map<String, String> params = new HashMap<String, String>(16);
       params.put(CommonParams.NAMESPACE_ID, namespaceId);
       params.put(CommonParams.SERVICE_NAME, serviceName);
       params.put(CommonParams.GROUP_NAME, groupName);
       params.put(CommonParams.CLUSTER_NAME, instance.getClusterName());
       params.put("ip", instance.getIp());
       params.put("port", String.valueOf(instance.getPort()));
       params.put("weight", String.valueOf(instance.getWeight()));
       params.put("enable", String.valueOf(instance.isEnabled()));
       params.put("healthy", String.valueOf(instance.isHealthy()));
       params.put("ephemeral", String.valueOf(instance.isEphemeral()));
       params.put("metadata", JacksonUtils.toJson(instance.getMetadata()));
       
       reqApi(UtilAndComs.nacosUrlInstance, params, HttpMethod.POST);
       
   }

服务端

@RestController
@RequestMapping(UtilsAndCommons.NACOS_NAMING_CONTEXT + “/instance”)
public class InstanceController //用于接收客户端注册

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

@CanDistro
   @PostMapping
   @Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
   public String register(HttpServletRequest request) throws Exception {
       
       final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
       final String namespaceId = WebUtils
               .optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
       
       final Instance instance = parseInstance(request);
       
       serviceManager.registerInstance(namespaceId, serviceName, instance);
       return "ok";
   }

订阅

1
2
3
4
5

@Override
 public void subscribe(String serviceName, EventListener listener) throws NacosException {
     subscribe(serviceName, new ArrayList<String>(), listener);
 }
1
2
3
4
@Override
   public void subscribe(String serviceName, List<String> clusters, EventListener listener) throws NacosException {
       subscribe(serviceName, Constants.DEFAULT_GROUP, clusters, listener);
   }
1
2
3
4
5
6
7
8

@Override
public void subscribe(String serviceName, String groupName, List<String> clusters, EventListener listener)
        throws NacosException {
    eventDispatcher.addListener(hostReactor
                    .getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ",")),
            StringUtils.join(clusters, ","), listener);
}

getServiceInfo

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
Hot Reactor
public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {
        
        NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch());
        String key = ServiceInfo.getKey(serviceName, clusters);
        if (failoverReactor.isFailoverSwitch()) {
            return failoverReactor.getService(key);
        }
        
        ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);
        
        if (null == serviceObj) {
            serviceObj = new ServiceInfo(serviceName, clusters);
            
            serviceInfoMap.put(serviceObj.getKey(), serviceObj);
            
            updatingMap.put(serviceName, new Object());
            updateServiceNow(serviceName, clusters);
            updatingMap.remove(serviceName);
            
        } else if (updatingMap.containsKey(serviceName)) {
            
            if (UPDATE_HOLD_INTERVAL > 0) {
                // hold a moment waiting for update finish
                synchronized (serviceObj) {
                    try {
                        serviceObj.wait(UPDATE_HOLD_INTERVAL);
                    } catch (InterruptedException e) {
                        NAMING_LOGGER
                                .error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e);
                    }
                }
            }
        }
        
        scheduleUpdateIfAbsent(serviceName, clusters);
        
        return serviceInfoMap.get(serviceObj.getKey());
    }


public void scheduleUpdateIfAbsent(String serviceName, String clusters) {
        if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
            return;
        }
        
        synchronized (futureMap) {
            if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
                return;
            }
            
            ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, clusters));
            futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);
        }
    }

UpdateTask

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

public class UpdateTask implements Runnable {
       
       long lastRefTime = Long.MAX_VALUE;
       
       private final String clusters;
       
       private final String serviceName;
       
       public UpdateTask(String serviceName, String clusters) {
           this.serviceName = serviceName;
           this.clusters = clusters;
       }
       
       @Override
       public void run() {
           long delayTime = -1;
           
           try {
               ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
               
               if (serviceObj == null) {
                   updateServiceNow(serviceName, clusters);
                   delayTime = DEFAULT_DELAY;
                   return;
               }
               
               if (serviceObj.getLastRefTime() <= lastRefTime) {
                   updateServiceNow(serviceName, clusters);
                   serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
               } else {
                   // if serviceName already updated by push, we should not override it
                   // since the push data may be different from pull through force push
                   refreshOnly(serviceName, clusters);
               }
               
               lastRefTime = serviceObj.getLastRefTime();
               
               if (!eventDispatcher.isSubscribed(serviceName, clusters) && !futureMap
                       .containsKey(ServiceInfo.getKey(serviceName, clusters))) {
                   // abort the update task
                   NAMING_LOGGER.info("update task is stopped, service:" + serviceName + ", clusters:" + clusters);
                   return;
               }
               
               delayTime = serviceObj.getCacheMillis();
               
           } catch (Throwable e) {
               NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e);
           } finally {
               if (delayTime > 0) {
                   executor.schedule(this, delayTime, TimeUnit.MILLISECONDS);
               }
           }
           
       }
   }

展开全文 >>

framework/SpringCloud/Nacos/Nacos概念名词

Nacos

https://nacos.io/zh-cn/docs/architecture.html

什么是 Nacos

Nacos 致力于帮助您发现、配置和管理微服务。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据及流量管理。

Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。 Nacos 是构建以“服务”为中心的现代应用架构 (例如微服务范式、云原生范式) 的服务基础设施。

服务（Service）是 Nacos 世界的一等公民。Nacos 支持几乎所有主流类型的“服务”的发现、配置和管理：

Nacos 的关键特性包括:

  • 服务发现和服务健康监测

    Nacos 支持基于 DNS 和基于 RPC 的服务发现。服务提供者使用 原生SDK、OpenAPI、或一个独立的Agent TODO注册 Service 后，服务消费者可以使用DNS TODO 或HTTP&API查找和发现服务。

    Nacos 提供对服务的实时的健康检查，阻止向不健康的主机或服务实例发送请求。Nacos 支持传输层 (PING 或 TCP)和应用层 (如 HTTP、MySQL、用户自定义）的健康检查。 对于复杂的云环境和网络拓扑环境中（如 VPC、边缘网络等）服务的健康检查，Nacos 提供了 agent 上报模式和服务端主动检测2种健康检查模式。Nacos 还提供了统一的健康检查仪表盘，帮助您根据健康状态管理服务的可用性及流量。

  • 动态配置服务

    动态配置服务可以让您以中心化、外部化和动态化的方式管理所有环境的应用配置和服务配置。

    动态配置消除了配置变更时重新部署应用和服务的需要，让配置管理变得更加高效和敏捷。

    配置中心化管理让实现无状态服务变得更简单，让服务按需弹性扩展变得更容易。

    Nacos 提供了一个简洁易用的UI (控制台样例 Demo) 帮助您管理所有的服务和应用的配置。Nacos 还提供包括配置版本跟踪、金丝雀发布、一键回滚配置以及客户端配置更新状态跟踪在内的一系列开箱即用的配置管理特性，帮助您更安全地在生产环境中管理配置变更和降低配置变更带来的风险。

  • 动态 DNS 服务

    动态 DNS 服务支持权重路由，让您更容易地实现中间层负载均衡、更灵活的路由策略、流量控制以及数据中心内网的简单DNS解析服务。动态DNS服务还能让您更容易地实现以 DNS 协议为基础的服务发现，以帮助您消除耦合到厂商私有服务发现 API 上的风险。

    Nacos 提供了一些简单的 DNS APIs TODO 帮助您管理服务的关联域名和可用的 IP:PORT 列表.

  • 服务及其元数据管理

    Nacos 能让您从微服务平台建设的视角管理数据中心的所有服务及元数据，包括管理服务的描述、生命周期、服务的静态依赖分析、服务的健康状态、服务的流量管理、路由及安全策略、服务的 SLA 以及最首要的 metrics 统计数据。

  • 更多的特性列表 …

里面的一些概念性名词

Region(地域)

Physical data centers, unalterable after resources are created.

物理的数据中心，资源创建成功后不能更换。

Available Zone（可用区）

Physical areas with independent power grids and networks in one region. The network latency for instances in the same zone is lower.

同一地域内，电力和网络互相独立的物理区域。同一可用区内，实例的网络延迟较低。

Endpoint（接入点）

The entry domain name of a service in each region

地域的某个服务的入口域名。

Namespace（命名空间）

For configuration isolation by tenants. Different namespaces may have configurations with the same Group or Data ID. One of the common scenarios for namespace is to differentiate and isolate the configurations in different environments, as in development and test environment and production environment.

用于进行租户粒度的配置隔离。不同的命名空间下，可以存在相同的 Group 或 Data ID 的配置。Namespace 的常用场景之一是不同环境的配置的区分隔离，例如开发测试环境和生产环境的资源（如配置、服务）隔离等。

Configuration配置

During system development, developers usually extract some parameters or variables that need to be changed from the code and manage them in a separate configuration file. This enables the static system artifacts or deliverables (such as WAR and JAR packages) to fit with the physical operating environment in a better way. Configuration management is usually a part of system deployment, which is executed by the administrator or operation and maintenance personnel,. Configuration modification is an effective way to adjust the behavior of a running system.

在系统开发过程中，开发者通常会将一些需要变更的参数、变量等从代码中分离出来独立管理，以独立的配置文件的形式存在。目的是让静态的系统工件或者交付物（如 WAR，JAR 包等）更好地和实际的物理运行环境进行适配。配置管理一般包含在系统部署的过程中，由系统管理员或者运维人员完成。配置变更是调整系统运行时的行为的有效手段。

Configuration Management（配置管理）

Configuration-related activities including editing, storage, distribution, modification management, release version management, and modification audit.

Configuration Item（配置项）

A specific configurable parameter with its value range, generally in the form of param-key=param-value. For example, the log output level (logLevel=INFO|WARN|ERROR) of a system is regarded as a configuration item.

一个具体的可配置的参数与其值域，通常以 param-key=param-value 的形式存在。例如我们常配置系统的日志输出级别（logLevel=INFO|WARN|ERROR） 就是一个配置项。

Configuration Set配置集

A collection of related or unrelated configuration items.In a system, a configuration file is generally a configuration set which contains all the configurations of the system. For example, a configuration set may contain configuration items such as data sources, thread pools, and log levels.

一组相关或者不相关的配置项的集合称为配置集。在系统中，一个配置文件通常就是一个配置集，包含了系统各个方面的配置。例如，一个配置集可能包含了数据源、线程池、日志级别等配置项。

Data ID（配置集 ID）

The ID of a configuration set in Nacos. It is one of the dimensions according to which configurations are organized. Data ID is generally used to organize the system configuration sets. A system or application can contain multiple configuration sets, each of which can be identified by a meaningful name. The Data ID usually uses the naming rule similar to Java packages (for example, com.taobao.tc.refund.log.level) to ensure global uniqueness. This naming rule is not mandatory.

Nacos 中的某个配置集的 ID。配置集 ID 是组织划分配置的维度之一。Data ID 通常用于组织划分系统的配置集。一个系统或者应用可以包含多个配置集，每个配置集都可以被一个有意义的名称标识。Data ID 通常采用类 Java 包（如 com.taobao.tc.refund.log.level）的命名规则保证全局唯一性。此命名规则非强制。

Group(配置分组)

The group of configuration sets in Nacos. It is one of the dimensions according to which configurations are organized. The configuration sets are always grouped by a meaningful string such as Buy or Trade to differentiate the configuration sets with the same Data ID. When you create a configuration on Nacos, the group name is replaced by DEFAULT_GROUP by default if not specified. A typical scenario of Group is when the same configuration type is used for different applications or components, such as database_url configuration and MQ_topic configuration.

Nacos 中的一组配置集，是组织配置的维度之一。通过一个有意义的字符串（如 Buy 或 Trade ）对配置集进行分组，从而区分 Data ID 相同的配置集。当您在 Nacos 上创建一个配置时，如果未填写配置分组的名称，则配置分组的名称默认采用 DEFAULT_GROUP 。配置分组的常见场景：不同的应用或组件使用了相同的配置类型，如 database_url 配置和 MQ_topic 配置。

Configuration Snapshot（配置快照）

The Nacos client SDK can generate snapshots of configurations on local machines. Snapshots can be used to indicate the overall disaster recovery capabilities of the system when the client cannot connect to the Nacos server. Configuration snapshot is similar to local commit in Git, or cache, which is updated at the appropriate time, but does not have the notion of expiration as in cache.

Nacos 的客户端 SDK 会在本地生成配置的快照。当客户端无法连接到 Nacos Server 时，可以使用配置快照显示系统的整体容灾能力。配置快照类似于 Git 中的本地 commit，也类似于缓存，会在适当的时机更新，但是并没有缓存过期（expiration）的概念。

Service（服务）

Software functions which are provided to the client via the network through a predefined interface.

通过预定义接口网络访问的提供给客户端的软件功能。

Service Name（服务名）

Identifier provided by the service, by which the service it refers to can be uniquely determined.

服务提供的标识，通过该标识可以唯一确定其指代的服务。

Service Registry（服务注册中心）

Database which stores the instances of services and the load balancing policies for services.

存储服务实例和服务负载均衡策略的数据库。

Service Discovery（服务发现）

On a computer network, the address and metadata of an instance under the service are probed (usually using a service name) and provided to the client for querying with a predefined interface.

在计算机网络上，（通常使用服务名）对服务下的实例的地址和元数据进行探测，并以预先定义的接口提供给客户端进行查询。

Metadata（元信息）

Custom configuration information, such as a disaster recovery policy, a load balancing policy, an authentication configuration, and various tags. From the scope of action, it is divided into meta-information of service level, meta-information of virtual cluster, and meta-information of instance.

Nacos数据（如配置和服务）描述信息，如服务版本、权重、容灾策略、负载均衡策略、鉴权配置、各种自定义标签 (label)，从作用范围来看，分为服务级别的元信息、集群的元信息及实例的元信息。

Application（应用）

Property of service which can be used to identify the service provider.

用于标识服务提供方的服务的属性。

Service Group（服务分组）

Different services can be categorized into the same service group.

不同的服务可以归类到同一分组。

Virtual Cluster（虚拟集群）

Service instances under the same service can be further classified. One possible unit of this classification is Virtual Cluster.

同一个服务下的所有服务实例组成一个默认集群, 集群可以被进一步按需求划分，划分的单位可以是虚拟集群。

Instance（实例）

A process with an accessible network address (IP:Port) that provides one or more services.

提供一个或多个服务的具有可访问网络地址（IP:Port）的进程。

Weight（权重）

Instance-level configuration. Weight is a floating-point number. The greater the weight, the greater the traffic that the instance expects to be allocated.

实例级别的配置。权重为浮点数。权重越大，分配给该实例的流量越大。

Health Check（健康检查）

Health check of the instances under a service in a specified manner to ensure that the instances can work properly. Instances are judged to be healthy or unhealthy according to the inspection results. Unhealthy instances are not returned to the client when initiating a resolution request to the service.

以指定方式检查服务下挂载的实例 (Instance) 的健康度，从而确认该实例 (Instance) 是否能提供服务。根据检查结果，实例 (Instance) 会被判断为健康或不健康。对服务发起解析请求时，不健康的实例 (Instance) 不会返回给客户端。

Protect Threshold（健康保护阈值）

To prevent traffic from flowing to healthy instances because of some unhealthy instances, which causes traffic pressure, healthy instance collapse, and finally an avalanche, the health protection threshold should be defined as a floating point number between 0 and 1. When the proportion of the domain name healthy instance to the total instance is smaller than this value, the instance is returned to the client regardless of the health of the instance. Although this can result in a loss of some of the traffic, we ensure that the remaining healthy instances can work normally.

为了防止因过多实例 (Instance) 不健康导致流量全部流向健康实例 (Instance) ，继而造成流量压力把健康 健康实例 (Instance) 压垮并形成雪崩效应，应将健康保护阈值定义为一个 0 到 1 之间的浮点数。当域名健康实例 (Instance) 占总服务实例 (Instance) 的比例小于该值时，无论实例 (Instance) 是否健康，都会将这个实例 (Instance) 返回给客户端。这样做虽然损失了一部分流量，但是保证了集群的剩余健康实例 (Instance) 能正常工作。

逻辑架构及其组件介绍

  • 服务管理：实现服务CRUD，域名CRUD，服务健康状态检查，服务权重管理等功能
  • 配置管理：实现配置管CRUD，版本管理，灰度管理，监听管理，推送轨迹，聚合数据等功能
  • 元数据管理：提供元数据CURD 和打标能力
  • 插件机制：实现三个模块可分可合能力，实现扩展点SPI机制
  • 事件机制：实现异步化事件通知，sdk数据变化异步通知等逻辑
  • 日志模块：管理日志分类，日志级别，日志可移植性（尤其避免冲突），日志格式，异常码+帮助文档
  • 回调机制：sdk通知数据，通过统一的模式回调用户处理。接口和数据结构需要具备可扩展性
  • 寻址模式：解决ip，域名，nameserver、广播等多种寻址模式，需要可扩展
  • 推送通道：解决server与存储、server间、server与sdk间推送性能问题
  • 容量管理：管理每个租户，分组下的容量，防止存储被写爆，影响服务可用性
  • 流量管理：按照租户，分组等多个维度对请求频率，长链接个数，报文大小，请求流控进行控制
  • 缓存机制：容灾目录，本地缓存，server缓存机制。容灾目录使用需要工具
  • 启动模式：按照单机模式，配置模式，服务模式，dns模式，或者all模式，启动不同的程序+UI
  • 一致性协议：解决不同数据，不同一致性要求情况下，不同一致性机制
  • 存储模块：解决数据持久化、非持久化存储，解决数据分片问题
  • Nameserver：解决namespace到clusterid的路由问题，解决用户环境与nacos物理环境映射问题
  • CMDB：解决元数据存储，与三方cmdb系统对接问题，解决应用，人，资源关系
  • Metrics：暴露标准metrics数据，方便与三方监控系统打通
  • Trace：暴露标准trace，方便与SLA系统打通，日志白平化，推送轨迹等能力，并且可以和计量计费系统打通
  • 接入管理：相当于阿里云开通服务，分配身份、容量、权限过程
  • 用户管理：解决用户管理，登录，sso等问题
  • 权限管理：解决身份识别，访问控制，角色管理等问题
  • 审计系统：扩展接口方便与不同公司审计系统打通
  • 通知系统：核心数据变更，或者操作，方便通过SMS系统打通，通知到对应人数据变更
  • OpenAPI：暴露标准Rest风格HTTP接口，简单易用，方便多语言集成
  • Console：易用控制台，做服务管理、配置管理等操作
  • SDK：多语言sdk
  • Agent：dns-f类似模式，或者与mesh等方案集成
  • CLI：命令行对产品进行轻量化管理，像git一样好用

展开全文 >>

framework/SpringCloud/Hystrix/hystrix

限流和熔断

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91

 private final HystrixCommandProperties properties;
    private final HystrixCommandMetrics metrics;

    /* track whether this circuit is open/closed at any given point in time (default to false==closed) */
    private AtomicBoolean circuitOpen = new AtomicBoolean(false);

    /* when the circuit was marked open or was last allowed to try a 'singleTest' */
    private AtomicLong circuitOpenedOrLastTestedTime = new AtomicLong();

    protected HystrixCircuitBreakerImpl(HystrixCommandKey key, HystrixCommandGroupKey commandGroup, HystrixCommandProperties properties, HystrixCommandMetrics metrics) {
        this.properties = properties;
        this.metrics = metrics;
    }

    public void markSuccess() {
        if (circuitOpen.get()) {
            if (circuitOpen.compareAndSet(true, false)) {
                //win the thread race to reset metrics
                //Unsubscribe from the current stream to reset the health counts stream.  This only affects the health counts view,
                //and all other metric consumers are unaffected by the reset
                metrics.resetStream();
            }
        }
    }

    @Override
    public boolean allowRequest() {
        if (properties.circuitBreakerForceOpen().get()) {
            // properties have asked us to force the circuit open so we will allow NO requests
            return false;
        }
        if (properties.circuitBreakerForceClosed().get()) {
            // we still want to allow isOpen() to perform it's calculations so we simulate normal behavior
            isOpen();
            // properties have asked us to ignore errors so we will ignore the results of isOpen and just allow all traffic through
            return true;
        }
        return !isOpen() || allowSingleTest();
    }

    public boolean allowSingleTest() {
        long timeCircuitOpenedOrWasLastTested = circuitOpenedOrLastTestedTime.get();
        // 1) if the circuit is open
        // 2) and it's been longer than 'sleepWindow' since we opened the circuit
        if (circuitOpen.get() && System.currentTimeMillis() > timeCircuitOpenedOrWasLastTested + properties.circuitBreakerSleepWindowInMilliseconds().get()) {
            // We push the 'circuitOpenedTime' ahead by 'sleepWindow' since we have allowed one request to try.
            // If it succeeds the circuit will be closed, otherwise another singleTest will be allowed at the end of the 'sleepWindow'.
            if (circuitOpenedOrLastTestedTime.compareAndSet(timeCircuitOpenedOrWasLastTested, System.currentTimeMillis())) {
                // if this returns true that means we set the time so we'll return true to allow the singleTest
                // if it returned false it means another thread raced us and allowed the singleTest before we did
                return true;
            }
        }
        return false;
    }

    @Override
    public boolean isOpen() {
        if (circuitOpen.get()) {
            // if we're open we immediately return true and don't bother attempting to 'close' ourself as that is left to allowSingleTest and a subsequent successful test to close
            return true;
        }

        // we're closed, so let's see if errors have made us so we should trip the circuit open
        HealthCounts health = metrics.getHealthCounts();

        // check if we are past the statisticalWindowVolumeThreshold
        if (health.getTotalRequests() < properties.circuitBreakerRequestVolumeThreshold().get()) {
            // we are not past the minimum volume threshold for the statisticalWindow so we'll return false immediately and not calculate anything
            return false;
        }

        if (health.getErrorPercentage() < properties.circuitBreakerErrorThresholdPercentage().get()) {
            return false;
        } else {
            // our failure rate is too high, trip the circuit
            if (circuitOpen.compareAndSet(false, true)) {
                // if the previousValue was false then we want to set the currentTime
                circuitOpenedOrLastTestedTime.set(System.currentTimeMillis());
                return true;
            } else {
                // How could previousValue be true? If another thread was going through this code at the same time a race-condition could have
                // caused another thread to set it to true already even though we were in the process of doing the same
                // In this case, we know the circuit is open, so let the other thread set the currentTime and report back that the circuit is open
                return true;
            }
        }
    }

}

隔离

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

private Observable<R> executeCommandWithSpecifiedIsolation(final AbstractCommand<R> _cmd) {
       if (properties.executionIsolationStrategy().get() == ExecutionIsolationStrategy.THREAD) {
           //线程的隔离
          
           return Observable.defer(new Func0<Observable<R>>() {
               @Override
               public Observable<R> call() {
                   executionResult = executionResult.setExecutionOccurred();
                   
                   
                   //多线程隔离，只允许一个线程往下执行
                   if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) {
                       return Observable.error(new IllegalStateException("execution attempted while in state : " + commandState.get().name()));
                   }

                   metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.THREAD);

                   if (isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT) {
                       //超时就抛出
                       return Observable.error(new RuntimeException("timed out before executing run()"));
                   }
                   
                   //状态设置
                   if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.STARTED)) {
                       //we have not been unsubscribed, so should proceed
                       //增加一个并发数 这里会记下并发数，用于外部 限流或隔离
                       HystrixCounters.incrementGlobalConcurrentThreads();
                       threadPool.markThreadExecution();
                       // store the command that is being run
                       endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
                       executionResult = executionResult.setExecutedInThread();
                       /**
                        * If any of these hooks throw an exception, then it appears as if the actual execution threw an error
                        */
                       try {
                           executionHook.onThreadStart(_cmd);
                           executionHook.onRunStart(_cmd);
                           executionHook.onExecutionStart(_cmd);
                           return getUserExecutionObservable(_cmd);
                       } catch (Throwable ex) {
                           return Observable.error(ex);
                       }
                   } else {
                       //已经取消订阅，立即返回
                       return Observable.error(new RuntimeException("unsubscribed before executing run()"));
                   }
               }
           }).doOnTerminate(new Action0() {
               
              
               @Override
               public void call() {
                   if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.TERMINAL)) {
                       handleThreadEnd(_cmd);
                   }
                   if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.TERMINAL)) {
                       //if it was never started and received terminal, then no need to clean up (I don't think this is possible)
                   }
                   //if it was unsubscribed, then other cleanup handled it
               }
           }).doOnUnsubscribe(new Action0() {
               @Override
               public void call() {
                   if (threadState.compareAndSet(ThreadState.STARTED, ThreadState.UNSUBSCRIBED)) {
                       handleThreadEnd(_cmd);
                   }
                   if (threadState.compareAndSet(ThreadState.NOT_USING_THREAD, ThreadState.UNSUBSCRIBED)) {
                       //if it was never started and was cancelled, then no need to clean up
                   }
                   //if it was terminal, then other cleanup handled it
               }
               
               //交给一个线程池执行
           }).subscribeOn(threadPool.getScheduler(new Func0<Boolean>() {
               @Override
               public Boolean call() {
                   
                   
                   return properties.executionIsolationThreadInterruptOnTimeout().get() && _cmd.isCommandTimedOut.get() == TimedOutStatus.TIMED_OUT;
               }
           }));
       } else {
           
           
           return Observable.defer(new Func0<Observable<R>>() {
               @Override
               public Observable<R> call() {
                   executionResult = executionResult.setExecutionOccurred();
                   
                   //线程隔离
                   if (!commandState.compareAndSet(CommandState.OBSERVABLE_CHAIN_CREATED, CommandState.USER_CODE_EXECUTED)) {
                       return Observable.error(new IllegalStateException("execution attempted while in state : " + commandState.get().name()));
                   }

                   metrics.markCommandStart(commandKey, threadPoolKey, ExecutionIsolationStrategy.SEMAPHORE);
                   // 信号量隔离
                   //存储正在运行的命令
                   
                   // //调用线程就是执行线程，这里会记下并发数，用于外部 限流或隔离
                   endCurrentThreadExecutingCommand = Hystrix.startCurrentThreadExecutingCommand(getCommandKey());
                   try {
                       executionHook.onRunStart(_cmd);
                       executionHook.onExecutionStart(_cmd);
                       return getUserExecutionObservable(_cmd);  //getUserExecutionObservable方法已经包装了同步异常，因此不应引发
                   } catch (Throwable ex) {
                       return Observable.error(ex);
                   }
               }
           });
       }
   }

展开全文 >>

framework/SpringCloud/Hystrix/SpringCloud的hystrix

hystrix集成到cloud

一个Observable一直包

Observable订阅是， 把新进来的订阅者 包成另一个 订阅者 床给 父的Observable

然后最后的架构就是

​ Observable —>Observable1->…–>Observablen

​ Subscribern <— Subscribern-1 <– Subscriber1

最后调用 Observable n的OnSubscribe ( Subscriber1 ) ->Observable n-1的OnSubscribe ( Subscriber2 ) —>Observable 1的OnSubscribe ( Subscriber n ) 

1
2
3
4
5
6
7
8
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.netflix.hystrix.HystrixAutoConfiguration,\
org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerAutoConfiguration,\
org.springframework.cloud.netflix.hystrix.ReactiveHystrixCircuitBreakerAutoConfiguration,\
org.springframework.cloud.netflix.hystrix.security.HystrixSecurityAutoConfiguration

org.springframework.cloud.client.circuitbreaker.EnableCircuitBreaker=\
org.springframework.cloud.netflix.hystrix.HystrixCircuitBreakerConfiguration

HystrixAutoConfiguration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
	@Bean
	@ConditionalOnEnabledHealthIndicator("hystrix")
	public HystrixHealthIndicator hystrixHealthIndicator() {
        //一个Hystrix健康指示器
		return new HystrixHealthIndicator();
	}

@Configuration(proxyBeanMethods = false)
	@ConditionalOnProperty(value = "management.metrics.binders.hystrix.enabled",
			matchIfMissing = true)
	@ConditionalOnClass({ HystrixMetricsBinder.class })
	protected static class HystrixMetricsConfiguration {

		@Bean
		public HystrixMetricsBinder hystrixMetricsBinder() {
            //指标绑定
			return new HystrixMetricsBinder();
		}

	}


@Configuration(proxyBeanMethods = false)
	@ConditionalOnWebApplication(type = SERVLET)
	@ConditionalOnBean(HystrixCommandAspect.class) // only install the stream if enabled
	@ConditionalOnClass({ HystrixMetricsStreamServlet.class })
	@EnableConfigurationProperties(HystrixProperties.class)
	protected static class HystrixServletAutoConfiguration {

        
        //提供一个servlet来供访问 服务的执行状态
		@Bean
		@ConditionalOnAvailableEndpoint
		public HystrixStreamEndpoint hystrixStreamEndpoint(HystrixProperties properties) {
			return new HystrixStreamEndpoint(properties.getConfig());
		}

		@Bean
		public HasFeatures hystrixStreamFeature() {
			return HasFeatures.namedFeature("Hystrix Stream Servlet",
					HystrixMetricsStreamServlet.class);
		}

	}

HystrixProperties

配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

@ConfigurationProperties(prefix = "management.endpoint.hystrix")
public class HystrixProperties {

	/**
	 * Hystrix settings. These are traditionally set using servlet parameters. Refer to
	 * the documentation of Hystrix for more details.
	 */
	private final Map<String, String> config = new HashMap<>();

	public Map<String, String> getConfig() {
		return this.config;
	}

}

HealthContributorAutoConfiguration

1
2
3
4
5
6
7
8
9
10
11
12
13
14

@Configuration(proxyBeanMethods = false)
public class HealthContributorAutoConfiguration {

	@Bean
	@ConditionalOnEnabledHealthIndicator("ping")
	public PingHealthIndicator pingHealthContributor() {
        
        //一个ping的健康指示器
        
		return new PingHealthIndicator();
	}

}

HystrixCircuitBreakerAutoConfiguration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


@Configuration(proxyBeanMethods = false)
@ConditionalOnClass({ Hystrix.class })
@ConditionalOnProperty(name = "spring.cloud.circuitbreaker.hystrix.enabled",
		matchIfMissing = true)
public class HystrixCircuitBreakerAutoConfiguration {

	@Autowired(required = false)
	private List<Customizer<HystrixCircuitBreakerFactory>> customizers = new ArrayList<>();

    
    
	@Bean
	@ConditionalOnMissingBean(CircuitBreakerFactory.class)
	public CircuitBreakerFactory hystrixCircuitBreakerFactory() {
        没有用到，不知道意义在哪
		HystrixCircuitBreakerFactory factory = new HystrixCircuitBreakerFactory();
		customizers.forEach(customizer -> customizer.customize(factory));
		return factory;
	}

}

HystrixCircuitBreaker

断路器

没有用到，不知道意义在哪

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public class HystrixCircuitBreaker implements CircuitBreaker {

	private HystrixCommand.Setter setter;

	public HystrixCircuitBreaker(HystrixCommand.Setter setter) {
		this.setter = setter;
	}

	@Override
	public <T> T run(Supplier<T> toRun, Function<Throwable, T> fallback) {

        
       //注意这里好像没有什么断路功能，但是断路功能HystrixCommand包含了，只要执行就能默认有断路功能
        //而我们需要关注的是HystrixCommand.Setter，这里面有我们需要配置的东西
		HystrixCommand<T> command = new HystrixCommand<T>(setter) {
			@Override
			protected T run() throws Exception {
				return toRun.get();
			}

			@Override
			protected T getFallback() {
				return fallback.apply(getExecutionException());
			}
		};
		return command.execute();
	}

}

HystrixCircuitBreakerConfiguration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

@Configuration(proxyBeanMethods = false)
public class HystrixCircuitBreakerConfiguration {

	@Bean
	public HystrixCommandAspect hystrixCommandAspect() {
		return new HystrixCommandAspect();
	}

	@Bean
	public HystrixShutdownHook hystrixShutdownHook() {
		return new HystrixShutdownHook();
	}

	@Bean
	public HasFeatures hystrixFeature() {
		return HasFeatures
				.namedFeatures(new NamedFeature("Hystrix", HystrixCommandAspect.class));
	}

	/**
	 * {@link DisposableBean} that makes sure that Hystrix internal state is cleared when
	 * {@link ApplicationContext} shuts down.
	 */
	private class HystrixShutdownHook implements DisposableBean {

		@Override
		public void destroy() throws Exception {
			// Just call Hystrix to reset thread pool etc.
			Hystrix.reset();
		}

	}

}

前面都是对其他框架的辅助，没卵用

重点执行

HystrixCommandAspect

切面

1
2
3
4
5
6
7
8
9
10
11
12
13

@RequestMapping(value = "/testHystrix")
@HystrixCommand(fallbackMethod = "helloFallback",commandKey = "queryAllKey")//熔断器指定操作的方法
public Object queryAll(){
    System.out.println("==========进入访问方法============");
    String url="aaa";
    List forObject = restTemplate.getForObject(url+"/queryAll", List.class);
    return forObject;
}

private Object helloFallback(){
    return null;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

   ----切点 @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCommand)")

    public void hystrixCommandAnnotationPointcut() {
    }

    @Pointcut("@annotation(com.netflix.hystrix.contrib.javanica.annotation.HystrixCollapser)")
    public void hystrixCollapserAnnotationPointcut() {
    }


切点会找到queryAll方法
    所以
    
    
 @Around("hystrixCommandAnnotationPointcut() || hystrixCollapserAnnotationPointcut()")
    public Object methodsAnnotatedWithHystrixCommand(final ProceedingJoinPoint joinPoint) throws Throwable {
    
    //获取执行的方法
        Method method = getMethodFromTarget(joinPoint);
        Validate.notNull(method, "failed to get method from joinPoint: %s", joinPoint);
        if (method.isAnnotationPresent(HystrixCommand.class) && method.isAnnotationPresent(HystrixCollapser.class)) {
            throw new IllegalStateException("method cannot be annotated with HystrixCommand and HystrixCollapser " +
                    "annotations at the same time");
        }
    
    //获取元数据持有者的工厂
    //只有两种
        MetaHolderFactory metaHolderFactory = META_HOLDER_FACTORY_MAP.get(HystrixPointcutType.of(method));
    
    //通过joinPoint来创建元数据持有者
        MetaHolder metaHolder = metaHolderFactory.create(joinPoint);
    
    //然后获取 持有调用器的对象
        HystrixInvokable invokable = HystrixCommandFactory.getInstance().create(metaHolder);
    //获取执行的类型
    //一般是没有Observable
        ExecutionType executionType = metaHolder.isCollapserAnnotationPresent() ?
                metaHolder.getCollapserExecutionType() : metaHolder.getExecutionType();

        Object result;
        try {
            if (!metaHolder.isObservable()) {
                
                //最后执行
                result = CommandExecutor.execute(invokable, executionType, metaHolder);
            } else {
                result = executeObservable(invokable, executionType, metaHolder);
            }
        } catch (HystrixBadRequestException e) {
            throw e.getCause();
        } catch (HystrixRuntimeException e) {
            throw hystrixRuntimeExceptionToThrowable(metaHolder, e);
        }
        return result;
    }

配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

private GenericSetterBuilder createGenericSetterBuilder(MetaHolder metaHolder) {
    GenericSetterBuilder.Builder setterBuilder = GenericSetterBuilder.builder()
            .groupKey(metaHolder.getCommandGroupKey())
            .threadPoolKey(metaHolder.getThreadPoolKey())
            .commandKey(metaHolder.getCommandKey())
            .collapserKey(metaHolder.getCollapserKey())
            .commandProperties(metaHolder.getCommandProperties())
            .threadPoolProperties(metaHolder.getThreadPoolProperties())
            .collapserProperties(metaHolder.getCollapserProperties());
    if (metaHolder.isCollapserAnnotationPresent()) {
        setterBuilder.scope(metaHolder.getHystrixCollapser().scope());
    }
    return setterBuilder.build();
}

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发
开源爱好者、Linux

联系QQ:2960571342

很惭愧

只做了一点微小的工作
谢谢大家