微服务架构：由浅入深带你了解底层注册中心

//1：服务之间通过RestTemplate调用，url写死
String url = "http://localhost:8020/order/findOrderByUserId/"+id;
User result = restTemplate.getForObject(url,User.class);

//2：类似还有其他http工具调用
String url = "http://localhost:8020/order/findOrderByUserId/" + id;
OkHttpClient client = new OkHttpClient();
Request request = new Request.Builder()
        .url(url)
        .build();
try (Response response = client.newCall(request).execute()) {
    String jsonResponse = response.body().string();
    // 处理 jsonResponse 对象。省略代码

//使用微服务名发起调用
String url = "http://mall‐order/order/findOrderByUserId/"+id;
List<Order> orderList = restTemplate.getForObject(url, List.class);

<dependencies>
    <dependency>
        <groupId>com.xiaoyu.mall</groupId>
        <artifactId>mall-common</artifactId>
        <version>0.0.1-SNAPSHOT</version>
        <scope>compile</scope>
    </dependency>

    <!-- nacos服务注册与发现 -->
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
    </dependency>
</dependencies>

server:
  port: 8040

spring:
  application:
    name: mall-user  #微服务名称

  #配置nacos注册中心地址
  cloud:
    nacos:
      discovery:
        server-addr: 127.0.0.1:8848
        namespace: 9f545878-ca6b-478d-8a5a-5321d58b3ca3

@Bean
@LoadBalanced  //mall-order => ip:port
public RestTemplate restTemplate() {
    return new RestTemplate();
}

@RequestMapping(value = "/findOrderByUserId/{id}")
public R  findOrderByUserId(@PathVariable("id") Integer id) {
    log.info("根据userId:"+id+"查询订单信息");
    // ribbon实现，restTemplate需要添加@LoadBalanced注解
    // mall-order  ip:port
    String url = "http://mall-order/order/findOrderByUserId/"+id;

    R result = restTemplate.getForObject(url,R.class);
    return result;
}

@RequestMapping("/findOrderByUserId/{userId}")
public R findOrderByUserId(@PathVariable("userId") Integer userId) {
    log.info("根据userId:"+userId+"查询订单信息");
    List<OrderEntity> orderEntities = orderService.listByUserId(userId);
    return R.ok().put("orders", orderEntities);
}

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-openfeign</artifactId>
</dependency>

@SpringBootApplication
@EnableFeignClients //扫描和注册feign客户端bean定义
public class MallUserFeignDemoApplication {、
  public static void main(String[] args) {
      SpringApplication.run(MallUserFeignDemoApplication.class, args);
  }
}

@FeignClient(value = "mall-order",path = "/order")
public interface OrderFeignService {
    @RequestMapping("/findOrderByUserId/{userId}")
    R findOrderByUserId(@PathVariable("userId") Integer userId);
}

@Autowired
OrderFeignService orderFeignService;

@RequestMapping(value = "/findOrderByUserId/{id}")
public R  findOrderByUserId(@PathVariable("id") Integer id) {
    //feign调用
    R result = orderFeignService.findOrderByUserId(id);
    return result;
}

@Override
public void register(Registration registration) {
  //此处省略非关键代码
    NamingService namingService = namingService();
    String serviceId = registration.getServiceId();
    String group = nacosDiscoveryProperties.getGroup();

    Instance instance = getNacosInstanceFromRegistration(registration);

    try {
        namingService.registerInstance(serviceId, group, instance);
        log.info("nacos registry, {} {} {}:{} register finished", group, serviceId,
                instance.getIp(), instance.getPort());
    }
    //此处省略非关键代码

private <T extends Response> T requestToServer(AbstractNamingRequest request, Class<T> responseClass)
        throws NacosException {
    try {
        request.putAllHeader(
                getSecurityHeaders(request.getNamespace(), request.getGroupName(), request.getServiceName()));
        Response response =
                requestTimeout < 0 ? rpcClient.request(request) : rpcClient.request(request, requestTimeout);
        //此处省略非关键代码

@Override
@TpsControl(pointName = "RemoteNamingServiceSubscribeUnSubscribe", name = "RemoteNamingServiceSubscribeUnsubscribe")
@Secured(action = ActionTypes.READ)
@ExtractorManager.Extractor(rpcExtractor = SubscribeServiceRequestParamExtractor.class)
public SubscribeServiceResponse handle(SubscribeServiceRequest request, RequestMeta meta) throws NacosException {
    String namespaceId = request.getNamespace();
    String serviceName = request.getServiceName();
    String groupName = request.getGroupName();
    String app = RequestContextHolder.getContext().getBasicContext().getApp();
    String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);
    Service service = Service.newService(namespaceId, groupName, serviceName, true);
    Subscriber subscriber = new Subscriber(meta.getClientIp(), meta.getClientVersion(), app, meta.getClientIp(),
            namespaceId, groupedServiceName, 0, request.getClusters());
    ServiceInfo serviceInfo = ServiceUtil.selectInstancesWithHealthyProtection(serviceStorage.getData(service),
            metadataManager.getServiceMetadata(service).orElse(null), subscriber.getCluster(), false, true,
            subscriber.getIp());
    if (request.isSubscribe()) {
        clientOperationService.subscribeService(service, subscriber, meta.getConnectionId());
        NotifyCenter.publishEvent(new SubscribeServiceTraceEvent(System.currentTimeMillis(),
                NamingRequestUtil.getSourceIpForGrpcRequest(meta), service.getNamespace(), service.getGroup(),
                service.getName()));
    } else {
        clientOperationService.unsubscribeService(service, subscriber, meta.getConnectionId());
        NotifyCenter.publishEvent(new UnsubscribeServiceTraceEvent(System.currentTimeMillis(),
                NamingRequestUtil.getSourceIpForGrpcRequest(meta), service.getNamespace(), service.getGroup(),
                service.getName()));
    }
    return new SubscribeServiceResponse(ResponseCode.SUCCESS.getCode(), "success", serviceInfo);
}

public final void start() throws NacosException {
       // 省略一些代码
        
        clientEventExecutor = new ScheduledThreadPoolExecutor(2, r -> {
            Thread t = new Thread(r);
            t.setName("com.alibaba.nacos.client.remote.worker");
            t.setDaemon(true);
            return t;
        });
        
        // 省略一些代码
        
        clientEventExecutor.submit(() -> {
            while (true) {
                try {
                    if (isShutdown()) {
                        break;
                    }
                    ReconnectContext reconnectContext = reconnectionSignal
                            .poll(keepAliveTime, TimeUnit.MILLISECONDS);
                    if (reconnectContext == null) {
                        // check alive time.
                        if (System.currentTimeMillis() - lastActiveTimeStamp >= keepAliveTime) {
                            boolean isHealthy = healthCheck();
                            if (!isHealthy) {
                                 // 省略一些代码

@Override
@TpsControl(pointName = "HealthCheck")
public HealthCheckResponse handle(HealthCheckRequest request, RequestMeta meta) {
    return new HealthCheckResponse();
}

Connection connection = connectionManager.getConnection(GrpcServerConstants.CONTEXT_KEY_CONN_ID.get());
RequestMeta requestMeta = new RequestMeta();
requestMeta.setClientIp(connection.getMetaInfo().getClientIp());
requestMeta.setConnectionId(GrpcServerConstants.CONTEXT_KEY_CONN_ID.get());
requestMeta.setClientVersion(connection.getMetaInfo().getVersion());
requestMeta.setLabels(connection.getMetaInfo().getLabels());
requestMeta.setAbilityTable(connection.getAbilityTable());
//这里刷新下时间。用来代表它确实存活
connectionManager.refreshActiveTime(requestMeta.getConnectionId());
prepareRequestContext(request, requestMeta, connection);
//这次处理的返回
Response response = requestHandler.handleRequest(request, requestMeta);

public void start() {
    initConnectionEjector();
    // Start UnHealthy Connection Expel Task.
    RpcScheduledExecutor.COMMON_SERVER_EXECUTOR.scheduleWithFixedDelay(() -> {
        runtimeConnectionEjector.doEject();
        MetricsMonitor.getLongConnectionMonitor().set(connections.size());
    }, 1000L, 3000L, TimeUnit.MILLISECONDS);
//省略部分代码，doEject方法再往后走，你就会发现这样一段代码
//outdated connections collect.
for (Map.Entry<String, Connection> entry : connections.entrySet()) {
    Connection client = entry.getValue();
    if (now - client.getMetaInfo().getLastActiveTime() >= KEEP_ALIVE_TIME) {
        outDatedConnections.add(client.getMetaInfo().getConnectionId());
    } else if (client.getMetaInfo().pushQueueBlockTimesLastOver(300 * 1000)) {
        outDatedConnections.add(client.getMetaInfo().getConnectionId());
    }
}
//省略部分代码，

public void run() {
    //省略部分代码
    if (serviceObj == null) {
        serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false);
        serviceInfoHolder.processServiceInfo(serviceObj);
        lastRefTime = serviceObj.getLastRefTime();
        return;
    }
    
    if (serviceObj.getLastRefTime() <= lastRefTime) {
        serviceObj = namingClientProxy.queryInstancesOfService(serviceName, groupName, clusters, 0, false);
        serviceInfoHolder.processServiceInfo(serviceObj);
    }
    //省略部分代码

public QueryServiceResponse handle(ServiceQueryRequest request, RequestMeta meta) throws NacosException {
    String namespaceId = request.getNamespace();
    String groupName = request.getGroupName();
    String serviceName = request.getServiceName();
    Service service = Service.newService(namespaceId, groupName, serviceName);
    String cluster = null == request.getCluster() ? "" : request.getCluster();
    boolean healthyOnly = request.isHealthyOnly();
    ServiceInfo result = serviceStorage.getData(service);
    ServiceMetadata serviceMetadata = metadataManager.getServiceMetadata(service).orElse(null);
    result = ServiceUtil.selectInstancesWithHealthyProtection(result, serviceMetadata, cluster, healthyOnly, true,
            NamingRequestUtil.getSourceIpForGrpcRequest(meta));
    return QueryServiceResponse.buildSuccessResponse(result);
}

public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint)
        throws IOException {
    ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
    Server server = getServer(loadBalancer, hint);
    if (server == null) {
        throw new IllegalStateException("No instances available for " + serviceId);
    }
    RibbonServer ribbonServer = new RibbonServer(serviceId, server,
            isSecure(server, serviceId),
            serverIntrospector(serviceId).getMetadata(server));

    return execute(serviceId, ribbonServer, request);
}

#被调用的微服务名
mall‐order:
 ribbon:
    #指定使用Nacos提供的负载均衡策略（优先调用同一集群的实例，基于随机&权重）
    NFLoadBalancerRuleClassName:com.alibaba.cloud.nacos.ribbon.NacosRule

@Bean
public IRule ribbonRule() {
    // 指定使用Nacos提供的负载均衡策略（优先调用同一集群的实例，基于随机权重）
    return new NacosRule();
}