Fork me on GitHub

framework/SpringCloud/Nacos/Nacos简单使用

字数统计: 1.7k字   |   阅读时长≈ 10分

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

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

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

很惭愧

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