Shulou(Shulou.com)06/03 Report--

In the last chapter, we learned about the module architecture of the Nacos project from the overall situation, and we have a good idea. Now, we are going to dig into the code details step by step. When many people are learning open source, they have no way to start, and there is so much code. Where should I start? We can start with an interface that you've used, and you know what it probably does, it's somatosensory, do you remember the HelloWorld we wrote in the first chapter, yes, start peeling onions from the interface inside.

Https://github.com/alibaba/nacos, this is the github code address of Nacos. Start follow it before you start, plus watch, and the subsequent mailing list of Nacos will also notify you, you can follow the latest real-time news of Nacos and wonderful discussions among the big bulls.

The following code is the code that publishes a service in the first chapter:

Public static void main (String [] args) throws NacosException, InterruptedException {/ / published service name String serviceName = "helloworld.services"; / / construct a Nacos instance. The input parameters are Nacos server ip and service port NamingService naming = NacosFactory.createNamingService ("100.81.0.34 NamingService naming 8080") / / publish a service with an ip of 127.0.0.1 and a port of 8888 naming.registerInstance (serviceName, "100.81.0.35", 8080); Thread.sleep (Integer.MAX_VALUE);}

The first step is to construct a Nacos service instance. The input parameter of the instance is a String, and the specification of the value is ip:port. This ip is the address of any of our Nacos server. Let's click on this method:

Public static NamingService createNamingService (String serverAddr) throws NacosException {return NamingFactory.createNamingService (serverAddr);}

At the same time, let's take a look at the code that creates the configuration service instance:

Public static ConfigService createConfigService (String serverAddr) throws NacosException {return ConfigFactory.createConfigService (serverAddr);}

We can see that NacosFactory is actually a unified entrance to the service discovery and configuration management interface, and then by its impassable method to create instances of different services, we can directly use NamingFactory, or ConfigFactory to create service instances of Nacos directly, or work

Cdn.nlark.com/lark/0/2018/png/4232/1537627839251-445ff4a4-d8ec-4cb0-abf5-ffd86d771bec.png ">

Next, take a look at how this Nacos naming instance is constructed:

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 (- 400, e.getMessage ());}}

An instance of NamingService, NacosNamingService, is instantiated through reflection, and the constructor is a String input. Let's look down and see what is done in the constructor:

Public NacosNamingService (String serverList) {this.serverList = serverList; init (); eventDispatcher = new EventDispatcher (); serverProxy = new NamingProxy (namespace, endpoint, serverList); beatReactor = new BeatReactor (serverProxy); hostReactor = new HostReactor (eventDispatcher, serverProxy, cacheDir);}

The input parameter serverList is the server address we just passed. The value is assigned to the serverList field of the instance, and then an init method is called, which is as follows:

Private void init () {namespace = System.getProperty (PropertyKeyConst.NAMESPACE); if (StringUtils.isEmpty (namespace)) {namespace = UtilAndComs.DEFAULT_NAMESPACE_ID;} logName = System.getProperty (UtilAndComs.NACOS_NAMING_LOG_NAME); if (StringUtils.isEmpty (logName)) {logName = "naming.log";} cacheDir = System.getProperty ("com.alibaba.nacos.naming.cache.dir") If (StringUtils.isEmpty (cacheDir)) {cacheDir = System.getProperty ("user.home") + "/ nacos/naming/" + namespace;}}

This side does three things: assign a value to namespace,logName,cacheDir. We have no input to namespace. By default, it is the role of default,namespace in Nacos, which is used for local cache isolation. On a machine, start a Nacos client process, and the default local cache path is default. If you start another one, you need to reset a namespace, otherwise the previous cache will be reused, resulting in conflicts. LogName and cacheDir, these two fields are not interpreted, literally. One more thing here, the setting of these values can be passed in the form of system parameters when java starts, and is the first priority.

After the init method is executed, the next step is to instantiate some framework components, EventDispatcher, a classic event distribution component that works as follows:

A separate thread will get the event from blockQueue. In Nacos, this event is the data pushed by the server. When we subscribe to a piece of data, listener will generate a listener instance, find the corresponding listener in the queue, and execute the callback function onEvent of listener in it. If you are not familiar with this model, you can look at the EventDispatcher code, which belongs to the basic knowledge, has nothing to do with the business, here is not too much detailed explanation, the length is too long.

Next, you instantiate a component of NameProxy. What does this thing do? Let's take a look at the code:

Public NamingProxy (String namespace, String endpoint, String serverList) {this.namespace = namespace; this.endpoint = endpoint; if (StringUtils.isNotEmpty (serverList)) {this.serverList = Arrays.asList (serverList.split (",")); if (this.serverList.size () = = 1) {this.nacosDomain = serverList } executorService = new ScheduledThreadPoolExecutor (1, new ThreadFactory () {@ Override public Thread newThread (Runnable r) {Thread t = newThread (r); t.setName ("com.taobao.vipserver.serverlist.updater"); t.setDaemon (true); return t;}}) ExecutorService.scheduleWithFixedDelay (new Runnable () {@ Override public void run () {refreshSrvIfNeed ();}}, 0, vipSrvRefInterMillis, TimeUnit.MILLISECONDS); refreshSrvIfNeed ();}

There is a lot of logic in this. To sum up, the main thing is to start a thread to execute the refreshSrvIfNeed () method every 30s.

In the refreshSrvIfNeed () method, what is done is to go to Nacos server to obtain a list of addresses of the Nacos server cluster through a http request, as shown below:

Private void refreshSrvIfNeed () {try {if (! CollectionUtils.isEmpty (serverList)) {LogUtils.LOG.info ("server list provided by user:" + serverList); return;} if (System.currentTimeMillis ()-lastSrvRefTime < vipSrvRefInterMillis) {return;} List list = getServerListFromEndpoint () If (list.isEmpty ()) {throw new Exception ("Can not acquire vipserver list");} if (! CollectionUtils.isEqualCollection (list, serversFromEndpoint)) {LogUtils.LOG.info ("SERVER-LIST", "server list is updated:" + list);} serversFromEndpoint = list; lastSrvRefTime = System.currentTimeMillis () } catch (Throwable e) {LogUtils.LOG.warn ("failed to update server list", e);}}

After the address list is obtained, it is assigned to serversFromEndpoint, and the current update time is recorded. When the next update is less than 30s, it will not be updated, avoiding frequent updates. Generally speaking, the purpose of NameProxy is to maintain the latest address list of Nacos server on the client side on a regular basis.

Let's move on, and then initialize the BeatReactor component, which can be guessed from the name, which should be related to the heartbeat. The initialization code is as follows:

Public BeatReactor (NamingProxy serverProxy) {this.serverProxy = serverProxy; executorService.scheduleAtFixedRate (new BeatProcessor (), 0, clientBeatInterval, TimeUnit.MILLISECONDS);}

A task with a fixed time interval of 10 seconds is performed to execute the logic in BeatProcessor. In the BeatProcessor code, the instance registered by the current client is cyclically fetched, and then a http heartbeat notification request is sent to the server to tell the client the health status of the service. The specific code is as follows:

Class BeatTask implements Runnable {BeatInfo beatInfo; public BeatTask (BeatInfo beatInfo) {this.beatInfo = beatInfo;} @ Override public void run () {Map params = new HashMap (2); params.put ("beat", JSON.toJSONString (beatInfo)); params.put ("dom", beatInfo.getDom ()) Try {String result = serverProxy.callAllServers (UtilAndComs.NACOS_URL_BASE + "/ api/clientBeat", params); JSONObject jsonObject = JSON.parseObject (result); if (jsonObject! = null) {clientBeatInterval = jsonObject.getLong ("clientBeatInterval") } catch (Exception e) {LogUtils.LOG.error ("CLIENT-BEAT", "failed to send beat:" + JSON.toJSONString (beatInfo), e);}

Here is the logic of the naocs client actively reporting the health status of the service. It is a more important concept of the service discovery function, the service health check mechanism, and commonly used by the server to actively detect the interface return of the client.

The final step is to initialize an instance called HostReactor. Let's take a look at what it does:

Public HostReactor (EventDispatcher eventDispatcher, NamingProxy serverProxy, String cacheDir) {this.eventDispatcher = eventDispatcher; this.serverProxy = serverProxy; this.cacheDir = cacheDir; this.serviceInfoMap = new ConcurrentHashMap (DiskCache.read (this.cacheDir)); this.failoverReactor = new FailoverReactor (this, cacheDir); this.pushRecver = new PushRecver (this);}

The fifth line is to load data from the cache file into the memory map of serviceInfoMap. Next, initialize a component of FailoverReactor, which is related to Nacos client cache disaster recovery. The initialization code in it is as follows:

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

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.