高考报名时间2022年11月1日还剩 天 距离2023年6月7日高考还有 天
栏目分类:
子分类:
返回
名师互学网用户登录
名师互学网
快速导航关闭
当前搜索
当前分类
前沿技术 软件开发 系统运维 产品运营 生活办公 面试经验 考试题库
子分类
人工智能 大数据 云计算 区块链 物联网 深度学习 机器学习 NLP 计算机视觉 语音识别 其他 大数据系统 数据可视化 数据挖掘与分析 其他 Docker/k8s 虚拟化 云平台 其他 基本原理 数字货币 智能合约 EOS应用 其他 通讯技术 嵌入式开发 单片机 物联网应用 HarmonyOS 其他 后端开发 Web开发 移动开发 游戏开发 Python Java 架构设计 C/C++/C# PHP .Net Go语言 R语言
实用工具
学习工具 小学数学练习 字帖生成 在线画板 函数绘制 拼音字母表 在线词典 黄历查询 亲戚关系计算 安全期计算 中国历史 Excel函数 模拟请求 json格式化 浏览器指纹
热门搜索
路由器设置 木托盘 宝塔面板 儿童python教程 心情低落 朋友圈 vim 双一流学科 专升本 我的学校 日记学校 西点培训学校 汽修学校 情书 化妆学校 塔沟武校 异形模板 西南大学排名 最精辟人生短句 6步教你追回被骗的钱 南昌大学排名 清朝十二帝 北京印刷学院排名 北方工业大学排名 北京航空航天大学排名 首都经济贸易大学排名 中国传媒大学排名 首都师范大学排名 中国地质大学(北京)排名 北京信息科技大学排名
名师互学网 > IT > 软件开发 > 后端开发 > Java

SpringCloud Nacos服务发现及调用过程(源码)

Java 更新时间： 发布时间： IT归档 最新发布 模块sitemap 名妆网 法律咨询 聚返吧 英语巴士网 伯小乐 网商动力

SpringCloud Nacos服务发现及调用过程(源码)

前提条件

客户端及服务端已经完成启动并注册成功

Feign远程调用

由于使用了feign, 所以向Spring容器中注入了FeignClientFactoryBean, 通过getObject方法, 获取到feign对应的代理对象, 对应的InvocationHandler为HystrixInvocationHandler. 当调用feign接口中方法时, 会执行HystrixInvocationHandler中的invoke方法.

public Object invoke(Object proxy, final Method method, final Object[] args) throws Throwable {
    if (!"equals".equals(method.getName())) {
        if ("hashCode".equals(method.getName())) {
            return this.hashCode();
        } else if ("toString".equals(method.getName())) {
            return this.toString();
        } else {
            HystrixCommand hystrixCommand = new HystrixCommand((Setter)this.setterMethodMap.get(method)) {
                protected Object run() throws Exception {
                    try {
                    	// 2.执行此方法, 开始调用
                        return ((MethodHandler)HystrixInvocationHandler.this.dispatch.get(method)).invoke(args);
                    } catch (Exception var2) {
                        throw var2;
                    } catch (Throwable var3) {
                        throw (Error)var3;
                    }
                }

                // 3.如果调用失败, 执行fallback
                protected Object getFallback() {
                	// 3.1.如果没有定义fallbackFactory, 则调用默认的, 抛出UnsupportedOperationException异常
                    if (HystrixInvocationHandler.this.fallbackFactory == null) {
                        return super.getFallback();
                    } else {
                        try {
                        	// 3.2.1.返回fallback对象
                            Object fallback = HystrixInvocationHandler.this.fallbackFactory.create(this.getExecutionException());
                            // 3.2.2.调用对应的方法
                            Object result = ((Method)HystrixInvocationHandler.this.fallbackMethodMap.get(method)).invoke(fallback, args);
                            if (HystrixInvocationHandler.this.isReturnsHystrixCommand(method)) {
                                return ((HystrixCommand)result).execute();
                            } else if (HystrixInvocationHandler.this.isReturnsObservable(method)) {
                                return ((Observable)result).toBlocking().first();
                            } else if (HystrixInvocationHandler.this.isReturnsSingle(method)) {
                                return ((Single)result).toObservable().toBlocking().first();
                            } else if (HystrixInvocationHandler.this.isReturnsCompletable(method)) {
                                ((Completable)result).await();
                                return null;
                            } else {
                            	// 3.2.3.返回结果
                                return HystrixInvocationHandler.this.isReturnsCompletableFuture(method) ? ((Future)result).get() : result;
                            }
                        } catch (IllegalAccessException var3) {
                            throw new AssertionError(var3);
                        } catch (ExecutionException | InvocationTargetException var4) {
                            throw new AssertionError(var4.getCause());
                        } catch (InterruptedException var5) {
                            Thread.currentThread().interrupt();
                            throw new AssertionError(var5.getCause());
                        }
                    }
                }
            };
            if (Util.isDefault(method)) {
                return hystrixCommand.execute();
            } else if (this.isReturnsHystrixCommand(method)) {
                return hystrixCommand;
            } else if (this.isReturnsObservable(method)) {
                return hystrixCommand.toObservable();
            } else if (this.isReturnsSingle(method)) {
                return hystrixCommand.toObservable().toSingle();
            } else if (this.isReturnsCompletable(method)) {
                return hystrixCommand.toObservable().toCompletable();
            } else {
            	// 1.首先执行此处, 调用hystrixCommand.execute(), 会执行hystrixCommand的run方法
                return this.isReturnsCompletableFuture(method) ? new ObservableCompletableFuture(hystrixCommand) : hystrixCommand.execute();
            }
        }
    } else {
        try {
            Object otherHandler = args.length > 0 && args[0] != null ? Proxy.getInvocationHandler(args[0]) : null;
            return this.equals(otherHandler);
        } catch (IllegalArgumentException var5) {
            return false;
        }
    }
}
 

SynchronousMethodHandler#invoke
 

public Object invoke(Object[] argv) throws Throwable {
	// 构造请求
    RequestTemplate template = buildTemplateFromArgs.create(argv);
    Request.Options options = findOptions(argv);
    Retryer retryer = this.retryer.clone();
    while (true) {
        try {
        	// 执行请求
            return executeAndDecode(template, options);
        } catch (RetryableException e) {
            try {
                retryer.continueOrPropagate(e);
            } catch (RetryableException th) {
                Throwable cause = th.getCause();
                if (propagationPolicy == UNWRAP && cause != null) {
                    throw cause;
                } else {
                    throw th;
                }
            }
            if (logLevel != Logger.Level.NONE) {
                logger.logRetry(metadata.configKey(), logLevel);
            }
            continue;
        }
    }
}


Object executeAndDecode(RequestTemplate template, Request.Options options) throws Throwable {
    Request request = targetRequest(template);

    if (logLevel != Logger.Level.NONE) {
        logger.logRequest(metadata.configKey(), logLevel, request);
    }

    Response response;
    long start = System.nanoTime();
    try {
    	// 执行 重点
        response = client.execute(request, options);
        // ensure the request is set. TODO: remove in Feign 12
        response = response.toBuilder()
                .request(request)
                .requestTemplate(template)
                .build();
    } catch (IOException e) {
        if (logLevel != Logger.Level.NONE) {
            logger.logIOException(metadata.configKey(), logLevel, e, elapsedTime(start));
        }
        throw errorExecuting(request, e);
    }
    long elapsedTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);


    if (decoder != null)
        return decoder.decode(response, metadata.returnType());

    CompletableFuture resultFuture = new CompletableFuture<>();
    // 处理结果
    asyncResponseHandler.handleResponse(resultFuture, metadata.configKey(), response,
            metadata.returnType(),
            elapsedTime);

    try {
        if (!resultFuture.isDone())
            throw new IllegalStateException("Response handling not done");

        return resultFuture.join();
    } catch (CompletionException e) {
        Throwable cause = e.getCause();
        if (cause != null)
            throw cause;
        throw e;
    }
}
 

LoadBalancerFeignClient#execute
 

public Response execute(Request request, Request.Options options) throws IOException {
	try {
		// 字符串转对象
		URI asUri = URI.create(request.url());
		String clientName = asUri.getHost();
		URI uriWithoutHost = cleanUrl(request.url(), clientName);
		FeignLoadBalancer.RibbonRequest ribbonRequest = new FeignLoadBalancer.RibbonRequest(
				this.delegate, request, uriWithoutHost);

		// ***** 此处重点 首次调用会更新服务列表 *****
		IClientConfig requestConfig = getClientConfig(options, clientName);
		// lbClient方法利用缓存创建并获取FeignLoadBalancer
		return lbClient(clientName)
				// 通过lb执行
				.executeWithLoadBalancer(ribbonRequest, requestConfig).toResponse();
	}
	catch (ClientException e) {
		IOException io = findIOException(e);
		if (io != null) {
			throw io;
		}
		throw new RuntimeException(e);
	}
}

 

AbstractLoadBalancerAwareClient#executeWithLoadBalancer
 

public T executeWithLoadBalancer(final S request, final IClientConfig requestConfig) throws ClientException {
	// 对象封装请求
    LoadBalancerCommand command = this.buildLoadBalancerCommand(request, requestConfig);

    try {
    	// submit中用来选取一个server然后调用call方法执行请求
        return (IResponse)command.submit(new ServerOperation() {
            public Observable call(Server server) {
                URI finalUri = AbstractLoadBalancerAwareClient.this.reconstructURIWithServer(server, request.getUri());
                ClientRequest requestForServer = request.replaceUri(finalUri);

                try {
                	// execute调用FeignLoadBalancer#execute方法 然后执行feign.Client.Default#execute 执行Http请求并构建相应
                    return Observable.just(AbstractLoadBalancerAwareClient.this.execute(requestForServer, requestConfig));
                } catch (Exception var5) {
                    return Observable.error(var5);
                }
            }
        }).toBlocking().single();
    } catch (Exception var6) {
        Throwable t = var6.getCause();
        if (t instanceof ClientException) {
            throw (ClientException)t;
        } else {
            throw new ClientException(var6);
        }
    }
}

 
服务发现 

LoadBalancerFeignClient#getClientConfig
 会调用NamedContextFactory#getContext根据Feign的name从Map中获取并缓存对应的context
 

protected AnnotationConfigApplicationContext getContext(String name) {
	if (!this.contexts.containsKey(name)) {
		synchronized (this.contexts) {
			if (!this.contexts.containsKey(name)) {
				// 只在首次调用会触发createContext
				this.contexts.put(name, createContext(name));
			}
		}
	}
	return this.contexts.get(name);
}

 

NamedContextFactory#createContext
 为当前各项配置封装成Spring容器, 并refresh容器, 每次创建一个新的context, 但是parentContext始终是同一个
 

protected AnnotationConfigApplicationContext createContext(String name) {
	AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
	if (this.configurations.containsKey(name)) {
		for (Class configuration : this.configurations.get(name)
				.getConfiguration()) {
			context.register(configuration);
		}
	}
	for (Map.Entry entry : this.configurations.entrySet()) {
		if (entry.getKey().startsWith("default.")) {
			for (Class configuration : entry.getValue().getConfiguration()) {
				context.register(configuration);
			}
		}
	}
	// defaultConfigType为org.springframework.cloud.netflix.ribbon.RibbonClientConfiguration为配置类
	context.register(PropertyPlaceholderAutoConfiguration.class,
			this.defaultConfigType);
	context.getEnvironment().getPropertySources().addFirst(new MapPropertySource(
			this.propertySourceName,
			Collections.singletonMap(this.propertyName, name)));
	// 设置父parent, 保证其他非feign的对象全局唯一
	if (this.parent != null) {
		// Uses Environment from parent as well as beans
		context.setParent(this.parent);
		// jdk11 issue
		// https://github.com/spring-cloud/spring-cloud-netflix/issues/3101
		context.setClassLoader(this.parent.getClassLoader());
	}
	context.setDisplayName(generateDisplayName(name));
	context.refresh();
	return context;
}

 

Spring加载配置类RibbonClientConfiguration后会初始化内部对象ribbonLoadBalancer, 调用父类DynamicServerListLoadBalancer的构造方法
 

public DynamicServerListLoadBalancer(IClientConfig clientConfig, IRule rule, IPing ping, ServerList serverList, ServerListFilter filter, ServerListUpdater serverListUpdater) {
    super(clientConfig, rule, ping);
    this.isSecure = false;
    this.useTunnel = false;
    this.serverListUpdateInProgress = new AtomicBoolean(false);
    this.updateAction = new NamelessClass_1();
    this.serverListImpl = serverList;
    this.filter = filter;
    this.serverListUpdater = serverListUpdater;
    if (filter instanceof AbstractServerListFilter) {
        ((AbstractServerListFilter)filter).setLoadBalancerStats(this.getLoadBalancerStats());
    }
    // 初始化
    this.restOfInit(clientConfig);
}

void restOfInit(IClientConfig clientConfig) {
    boolean primeConnection = this.isEnablePrimingConnections();
    this.setEnablePrimingConnections(false);
    this.enableAndInitLearnNewServersFeature();
    // 更新服务list
    this.updateListOfServers();
    if (primeConnection && this.getPrimeConnections() != null) {
        this.getPrimeConnections().primeConnections(this.getReachableServers());
    }

    this.setEnablePrimingConnections(primeConnection);
    LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());
}

public void updateListOfServers() {
    List servers = new ArrayList();
    if (this.serverListImpl != null) {
    	// 从远端获取对应的服务列表 会调用HostReactor#getServiceInfo方法
        servers = this.serverListImpl.getUpdatedListOfServers();
        if (this.filter != null) {
            servers = this.filter.getFilteredListOfServers((List)servers);
        }
    }
    // 将server通过zone分开存放成map
    this.updateAllServerList((List)servers);
}

public void setServersList(List lsrv) {
    super.setServersList(lsrv);
    List serverList = (List) lsrv;
    Map> serversInZones = new HashMap>();
    for (Server server : serverList) {
        // make sure ServerStats is created to avoid creating them on hot
        // path
        getLoadBalancerStats().getSingleServerStat(server);
        String zone = server.getZone(); // 默认UNKNOWN
        // 分类存放
        if (zone != null) {
            zone = zone.toLowerCase();
            List servers = serversInZones.get(zone);
            if (servers == null) {
                servers = new ArrayList();
                serversInZones.put(zone, servers);
            }
            servers.add(server);
        }
    }
    setServerListForZones(serversInZones);
}

 

HostReactor#getServiceInfo
 

public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {
    String key = ServiceInfo.getKey(serviceName, clusters);
    if (failoverReactor.isFailoverSwitch()) {
        return failoverReactor.getService(key);
    }

    // 从serviceInfoMap缓存中获取 key: name + "@@" + clusters, value: ServiceInfo包含服务列表
    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) { // 休息5s
            // 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);
                }
            }
        }
    }
    // 与心跳类似, 间隔10s更新服务列表 调用addTask添加定时
    scheduleUpdateIfAbsent(serviceName, clusters);
    return serviceInfoMap.get(serviceObj.getKey());
}

public void updateServiceNow(String serviceName, String clusters) {
    ServiceInfo oldService = getServiceInfo0(serviceName, clusters);
    try {
    	// 发起远程调用
        String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUDPPort(), false);
        if (StringUtils.isNotEmpty(result)) {
        	// 结果转换, 并存放到map缓存中
            processServiceJSON(result);
        }
    } catch (Exception e) {
        NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, e);
    } finally {
        if (oldService != null) {
            synchronized (oldService) {
                oldService.notifyAll();
            }
        }
    }
}

 

最终调用NamingProxy#queryList
 

public String queryList(String serviceName, String clusters, int udpPort, boolean healthyOnly)
    throws NacosException {

    final Map params = new HashMap(8);
    params.put(CommonParams.NAMESPACE_ID, namespaceId);
    params.put(CommonParams.SERVICE_NAME, serviceName);
    params.put("clusters", clusters);
    params.put("udpPort", String.valueOf(udpPort));
    params.put("clientIP", NetUtils.localIP());
    params.put("healthyOnly", String.valueOf(healthyOnly));

    return reqAPI(UtilAndComs.NACOS_URL_base + "/instance/list", params, HttpMethod.GET);
}

 

UpdateTask实现了Runnable接口, 执行run方法
 

public void run() {
    try {
        ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
        if (serviceObj == null) {
            updateServiceNow(serviceName, clusters);
            executor.schedule(this, DEFAULT_DELAY, TimeUnit.MILLISECONDS);
            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;
        }
        // 再次注册定时任务 循环调用
        executor.schedule(this, serviceObj.getCacheMillis(), TimeUnit.MILLISECONDS);
    } catch (Throwable e) {
        NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e);
    }
}

 
服务列表结果示例 

{
    "name": "DEFAULT_GROUP@@order-server", 
    "groupName": "DEFAULT_GROUP", 
    "clusters": "", 
    "cacheMillis": 10000, 
    "hosts": [
        {
            "instanceId": "10.247.36.102#8081#DEFAULT#DEFAULT_GROUP@@order-server", 
            "ip": "10.247.36.102", 
            "port": 8081, 
            "weight": 1, 
            "healthy": true, 
            "enabled": true, 
            "ephemeral": true, 
            "clusterName": "DEFAULT", 
            "serviceName": "DEFAULT_GROUP@@order-server", 
            "metadata": {
                "preserved.register.source": "SPRING_CLOUD"
            }, 
            "instanceHeartBeatTimeOut": 15000, 
            "ipDeleteTimeout": 30000, 
            "instanceHeartBeatInterval": 5000
        }
    ], 
    "lastRefTime": 1636438205772, 
    "checksum": "", 
    "allIPs": false, 
    "reachProtectionThreshold": false, 
    "valid": true
}








转载请注明：文章转载自 www.mshxw.com















Java相关栏目本月热门文章












热门相关搜索






路由器设置
木托盘
宝塔面板
儿童python教程
心情低落
朋友圈
vim
双一流学科
专升本
我的学校
日记学校
西点培训学校
汽修学校
情书
化妆学校
塔沟武校
异形模板
西南大学排名
最精辟人生短句
6步教你追回被骗的钱
南昌大学排名
清朝十二帝
北京印刷学院排名
北方工业大学排名
北京航空航天大学排名
首都经济贸易大学排名
中国传媒大学排名
首都师范大学排名
中国地质大学(北京)排名
北京信息科技大学排名
中央民族大学排名
北京舞蹈学院排名
北京电影学院排名
中国戏曲学院排名
河北政法职业学院排名
河北经贸大学排名
天津中德应用技术大学排名
天津医学高等专科学校排名
天津美术学院排名
天津音乐学院排名
天津工业大学排名
北京工业大学耿丹学院排名
北京警察学院排名
天津科技大学排名
北京邮电大学(宏福校区)排名
北京网络职业学院排名
北京大学医学部排名
河北科技大学排名
河北地质大学排名
河北体育学院排名















学习工具


代数计算器


三角函数


解析几何


立体几何






知识解答


教育知识


百科知识


生活知识


常识知识






写作必备


作文大全


作文素材


句子大全



实用范文






关于我们


关于我们


联系我们


网站地图






 交流群

名师互学网交流群




名师互学网客服

名师互学网客服












名师互学网 版权所有 (c)2021-2022      ICP备案号：晋ICP备2021003244-6号
 








我们一直用心在做

关于我们
文章归档
网站地图
联系我们

版权所有 (c)2021-2022 MSHXW.COM


ICP备案号：晋ICP备2021003244-6号