ServiceComb场景及其原理

Java-Chassis核心源码

Dubbo3场景及其原理
ServiceComb中的Java-Chassis模块是实现类似Dubbo的RPC的框架，又称Java底座，Dubbo解析请参考上一篇博文。

Consumer到Provider整体流程

• Consumer和Provider的Handler都支持通过SPI动态扩展，如流控、熔断、隔离、负载均衡等扩展。
• Consumer在传输层支持Filter拦截过滤请求，支持通过扩展点处理Request和Response。
• 监听者模式的完美实现，通过SPI读取监听者，监听者监听控制中心的状态，修改控制中心的状态信息。
• 自己业务代码也可以使用这种方式扩展功能并动态组合策略，解耦各种功能业务代码。

@EnableServiceComb-初始化ServiceComb

通过@ImportResource的方式导入ServiceComb的核心Bean定义文件"classpath*:META-INF/spring/scb-core-bean.xml";和"classpath*:META-INF/spring/*.bean.xml";，注入如下Bean定义。

• ConfigurationSpringInitializer

• CseApplicationListener，ContextRefreshedEvent事件监听器

  • 在CseApplicationListener#setApplicationContext初始化相关的连接

    if (this.applicationContext == applicationContext) {
          // same object. avoid initialize many times.
          return;
        }
        this.applicationContext = applicationContext;
        BeanUtils.setContext(applicationContext);
        HttpClients.load(); // Http相关的
        RegistrationManager.INSTANCE.init(); // 服务注册 管理初始化，通过SPI获取。
        DiscoveryManager.INSTANCE.init(); // 服务发现 管理初始化，通过SPI获取。
    1
    2
    3
    4
    5
    6
    7
    8
    9

  • ContextRefreshedEvent Bean实例化完事件触达，初始化ServiceComb引擎SCBEngine

      @Override
      public void onApplicationEvent(ApplicationEvent event) {
        if (initEventClass.isInstance(event)) {
          if (applicationContext instanceof AbstractApplicationContext) {
            ((AbstractApplicationContext) applicationContext).registerShutdownHook();
          }
    
          SCBEngine scbEngine = SCBEngine.getInstance(); // 1、获取单例
          
          scbEngine.setApplicationContext(applicationContext); // 2、设置Contex
          scbEngine.setPriorityPropertyManager(applicationContext.getBean(PriorityPropertyManager.class));
          scbEngine.setFilterChainsManager(applicationContext.getBean(FilterChainsManager.class)); // 3、指责链管理类
          scbEngine.getConsumerProviderManager().getConsumerProviderList()
              .addAll(applicationContext.getBeansOfType(ConsumerProvider.class).values()); // 4、获取支持的Consumer形式，如Rest/Pojo
          scbEngine.getProducerProviderManager().getProducerProviderList()
              .addAll(applicationContext.getBeansOfType(ProducerProvider.class).values()); // // 5、获取Provider注册的元信息，支持从Rest/Pojo中获取
          scbEngine.addBootListeners(applicationContext.getBeansOfType(BootListener.class).values()); // 6、获取全量SCB初始化监听器BootListener，用于监听SCB在生命周期内的各种事件。
    
          scbEngine.run(); // 7、run方法
        } else if (event instanceof ContextClosedEvent) {
          if (SCBEngine.getInstance() != null) {
            SCBEngine.getInstance().destroy();
          }
        }
      }
    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

  1. 获取单例SCBEngine<重载类，必须单例，Double Check，骚操作可实现隔离>，会首先初始化事件总线等逻辑。

     public static SCBEngine getInstance() {
         if (null == INSTANCE) {
           // private static final Object initializationLock = new Object(); 全局锁
           synchronized (initializationLock) {
             if (null == INSTANCE) {
               new SCBEngine();
             }
           }
         }
         return INSTANCE;
       }
     
     protected SCBEngine() {
         eventBus = EventManager.getEventBus(); // 获取事件总线
     
         eventBus.register(this); // 注册事件 @Subscribe  @AllowConcurrentEvents
     
         INSTANCE = this;
     
         producerProviderManager = new ProducerProviderManager(this);
         serviceRegistryListener = new ServiceRegistryListener(this);
       }
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22

     • 初始化事件总线，使用Guava事件总线处理事件，解耦内部各个模块的逻辑。

     • ProducerProviderManager生产提供者管理器，用于管理不同的Provider类型，包括PojoProducerProvider、RestProducerProvider，下文会通过注解的方式讲解这两种Provider。

     • SCBEngine监听InvocationStartEvent/InvocationFinishEvent事件，用于AtomicLong统计调用开始和调用结束的总数。

         @AllowConcurrentEvents
         @Subscribe
         public void onInvocationStart(InvocationStartEvent event) {
           invocationStartedCounter.incrementAndGet();
         }
       
         @AllowConcurrentEvents
         @Subscribe
         public void onInvocationFinish(InvocationFinishEvent event) {
           invocationFinishedCounter.incrementAndGet();
         }
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11

     • 注册ServiceRegistryListener到事件总线，监听以下事件，将引擎需要关注的，和服务发现类的各项操作解耦开，这是常规的操作，我们一般还是可以效仿的。

         @EnableExceptionPropagation
         @SubscriberOrder(-1000)
         @Subscribe
         public void onCreateMicroservice(CreateMicroserviceEvent event);
         
         @EnableExceptionPropagation
         @SubscriberOrder(-1000)
         @Subscribe
         public void onDestroyMicroservice(DestroyMicroserviceEvent event);
         
         @EnableExceptionPropagation
         @SubscriberOrder(-1000)
         @Subscribe
         public void onCreateMicroserviceVersion(CreateMicroserviceVersionEvent event);
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14

  2. 初始化消费者管理，提供者管理，BootListener，BootListener支持SPI或者@Component注解扩展，这种事件总线的模式值得产参考参考各位。

     • 监听各种事件，如切handler或filter前后、服务注册前后、传输前后、消费和生产提供者前后等，支持如下事件

       enum EventType {
           BEFORE_HANDLER,
           AFTER_HANDLER,
           BEFORE_FILTER,
           AFTER_FILTER,
           BEFORE_PRODUCER_PROVIDER,
           AFTER_PRODUCER_PROVIDER,
           BEFORE_CONSUMER_PROVIDER,
           AFTER_CONSUMER_PROVIDER,
           BEFORE_TRANSPORT,
           AFTER_TRANSPORT,
           BEFORE_REGISTRY,
           AFTER_REGISTRY,
           BEFORE_CLOSE,
           AFTER_CLOSE
         }
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16

  3. scb.run -> doRun方法，可以打开org.apache.servicecomb.core.SCBEngine的日志

     public synchronized SCBEngine run() {
         if (SCBStatus.DOWN.equals(status)) {
           try {
             doRun(); 
             waitStatusUp(); // 等待SCB变成UP状态，会在收到MicroserviceInstanceRegisteredEvent事件的时候修改。     
           } catch (TimeoutException e) {
           } catch (Throwable e) {
             try {
               destroy();
             } catch (Exception exception) {
             }
             status = SCBStatus.FAILED;
           } finally {
             printServiceInfo(); // 最终打印Service的全部信息
           }
         }
     private void doRun() throws Exception {
         status = SCBStatus.STARTING;
     
         // 1、BootListener排序包括SPI+Compnent扩展的类
         bootListeners.sort(Comparator.comparingInt(BootListener::getOrder));
     
         // 2、触发BootEvent调用BootListener里面对应的监听方法,上下文Contex类，不论是监听者还是职责链，都会需要传递上下文的Contex类。
         triggerEvent(EventType.BEFORE_HANDLER);
         
         // 3、通过SPI加载全部的Handler到Config.class中，并初始化到Manager中
         HandlerConfigUtils.init(consumerHandlerManager, producerHandlerManager);
         
         // 4、触发Handler加载之后的事件,我们可以监听handler加载到Engine的事件，这种写法爽不，非常爽，事件总线。
         triggerEvent(EventType.AFTER_HANDLER);
         
         // 5、同理操作Filter
         triggerEvent(EventType.BEFORE_FILTER);
         filterChainsManager.init();
         triggerEvent(EventType.AFTER_FILTER);
         
     	// 6、创建提供者元信息
         createProducerMicroserviceMeta();
     
         // 7、初始化生产管理者并触发事件
         triggerEvent(EventType.BEFORE_PRODUCER_PROVIDER);
         producerProviderManager.init();
         triggerEvent(EventType.AFTER_PRODUCER_PROVIDER);
     
         // 8、初始化消费管理者并触发事件
         triggerEvent(EventType.BEFORE_CONSUMER_PROVIDER);
         consumerProviderManager.init(); // 含有生命周期的Class可以使用init，destory等模板方法
           /**
           for (ConsumerProvider provider : consumerProviderList) {
           		provider.init(); // 模板方法
         	}
           **/
         triggerEvent(EventType.AFTER_CONSUMER_PROVIDER);
     
         // 9、初始化TransportManager管理器，通过SPI加载Transport的实现，
         // 包括HighwayTransport/ServletRestTransport/VertxRestTransport
         // Transport的实现都有名字，同名会选择其中的一个。
         // 加载选择完会调用Transport的Init方法初始化。
         triggerEvent(EventType.BEFORE_TRANSPORT);
         transportManager.init(this);
         triggerEvent(EventType.AFTER_TRANSPORT);
     
         triggerEvent(EventType.BEFORE_REGISTRY);
     
         // 10、注册AfterRegistryEventHanlder监MicroserviceInstanceRegisteredEvent事件，修改scb的状态为up
         triggerAfterRegistryEvent(); 
     
         // 11、服务注册启动
         RegistrationManager.INSTANCE.run(); 
         // 12、服务发现启动
         DiscoveryManager.INSTANCE.run(); 
     
         shutdownHook = new Thread(this::destroyForShutdownHook);
         Runtime.getRuntime().addShutdownHook(shutdownHook);
       }    
         
     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

     通过以上注释的数字索引详解部分有价值的地方。

     • 9、TransportManager

       public void init(SCBEngine scbEngine) throws Exception {
           buildTransportMap(); // SPI分组并选择
       
           for (Transport transport : transportMap.values()) {
             if (transport.init()) { // 初始化，初始化失败会出问题的
               Endpoint endpoint = transport.getPublishEndpoint();
               if (endpoint != null && endpoint.getEndpoint() != null) {
                 LOGGER.info("endpoint to publish: {}", endpoint.getEndpoint());
                 RegistrationManager.INSTANCE.addEndpoint(endpoint.getEndpoint()); // 注册到注册中心中
               }
               continue;
             }
           }
         }
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14

     • 11、RegistrationManager，非常重要，见下详解。

     • 12、DiscoveryManager

TransportManager-管理传输层

通过SPI解耦了各个Module，用户也可以基于实现自定义Transport层的实现，这种策略方式还是可以学习的。

1. java-chassis-core-2.6.0.jar核心包中提供了TransportManager、AbstractTransport、Transport基础类。TransportManager通过SPI加载Transport.class的实现类，提供了AbstractTransport.class用于将传输层的功能抽象到父类中。
2. transport-rest-vertx/transport-rest-servlet/transport-highway三个都是独立的jar包，用于扩展Transport

TransportManager-管理传输层

通过SPI解耦了各个Module，用户也可以基于实现自定义Transport层的实现，这种策略方式还是可以学习的。

1. java-chassis-core-2.6.0.jar核心包中提供了TransportManager、AbstractTransport、Transport基础类。TransportManager通过SPI加载Transport.class的实现类，提供了AbstractTransport.class用于将传输层的功能抽象到父类中。

2. transport-rest-vertx/transport-rest-servlet/transport-highway三个都是独立的jar包，用于扩展Transport。

3. Rest模式的transport会经过各种HttpClientFilter来动态拦截前后的请求，在拦截流程中通过Invocation这个Contex上下文来传递全量信息，这种思路非常的通用极易学习，代码这样写起来非常的舒服好看，这里的异步编程也是值得我们学习的，后面可以细看。

   org.apache.servicecomb.transport.rest.client.http.RestClientInvocation#invoke
   public void invoke(Invocation invocation, AsyncResponse asyncResp) throws Exception {
       this.invocation = invocation;
       this.asyncResp = asyncResp;
   
       OperationMeta operationMeta = invocation.getOperationMeta();
       restOperationMeta = operationMeta.getExtData(RestConst.SWAGGER_REST_OPERATION);
   
       String path = this.createRequestPath(restOperationMeta);
       IpPort ipPort = (IpPort) invocation.getEndpoint().getAddress();
   
       Future<HttpClientRequest> requestFuture = createRequest(ipPort, path);
   
       invocation.getInvocationStageTrace().startGetConnection();
       requestFuture.compose(clientRequest -> {
         invocation.getInvocationStageTrace().finishGetConnection();
   
         this.clientRequest = clientRequest;
   
         clientRequest.putHeader(org.apache.servicecomb.core.Const.TARGET_MICROSERVICE, invocation.getMicroserviceName());
         RestClientRequestImpl restClientRequest =
             new RestClientRequestImpl(clientRequest, httpClientWithContext.context(), asyncResp, throwableHandler);
         invocation.getHandlerContext().put(RestConst.INVOCATION_HANDLER_REQUESTCLIENT, restClientRequest);
   
         Buffer requestBodyBuffer;
         try {
           requestBodyBuffer = restClientRequest.getBodyBuffer();
         } catch (Exception e) {
           return Future.failedFuture(e);
         }
         HttpServletRequestEx requestEx = new VertxClientRequestToHttpServletRequest(clientRequest, requestBodyBuffer);
         invocation.getInvocationStageTrace().startClientFiltersRequest();
         // 1、执行Filter.beforeSendRequest，可用于动态修改request撒  
         for (HttpClientFilter filter : httpClientFilters) {
           if (filter.enabled()) {
             filter.beforeSendRequest(invocation, requestEx);
           }
         }
   
         // 2、通过事件总线发事件撒
         invocation.onStartSendRequest();
         httpClientWithContext.runOnContext(httpClient -> {
           clientRequest.setTimeout(operationMeta.getConfig().getMsRequestTimeout());
           clientRequest.response().onComplete(asyncResult -> {
             if (asyncResult.failed()) {
               fail(asyncResult.cause());
               return;
             }
             // 3、执行Filter.afterReceiveResponse，可用于动态修改Response撒  
             handleResponse(asyncResult.result());
           });
           processServiceCombHeaders(invocation, operationMeta);
           restClientRequest.end()
               .onComplete((t) -> invocation.getInvocationStageTrace().finishWriteToBuffer(System.nanoTime()));
         });
         return Future.succeededFuture();
       }).onFailure(failure -> {
       });
     }
   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

RegistrationManager-注册服务

Faced门面类，用于管理全部的Registration，功能全部路由到下游实体类，Registration是实际管理Provider注册到注册中心的逻辑(支持动态扩展不同的注册中心，可插拔)。会通过SPI的方式加载全部的Registration.class实现，解耦和实际注册中心处理类的实现，Faced设计模式可以学习，门面简化内部的实现细节及其使用。

1. RegistrationManager.INSTANCE.init()INSTANCE单例初始化，模板方法addBeans有些牛逼，做了两件事，非常值得学习啊。
   其一：自动给SPI加载的类Autowire(兼容Spring)。

   其二：将Spring框架里面的Bean也加载到RegistrationManager之中。

   // 1、SPI加载全部的实现，并调用过滤方法
   SPIServiceUtils.getOrLoadSortedService(Registration.class)
           .stream()
           .filter((SPIEnabled::enabled))
           .forEach(registrationList::add);
   // 2、会在CseApplicationListener#setApplicationContext被调用的init
   public void init() {
       //
       BeanUtils.addBeans(Registration.class, registrationList);
   
       initPrimary();
       // 初始化全部的registration
       registrationList.forEach(registration -> registration.init());
     }
   
   
   public static <T extends SPIOrder & SPIEnabled> void addBeans(Class<T> cls, List<T> exists) {
       if (context == null) {
         return;
       }
       
   	// 尝试通过Spring中的Bean autowire相应的instance
       for (T instance : exists) {
         context.getAutowireCapableBeanFactory().autowireBean(instance);
       }
       // 添加到exists中
       for (T bean : context.getBeansOfType(cls).values()) {
         if (bean.enabled()) {
           exists.add(bean);
         }
       }
       // 排个序呗
       exists.sort(Comparator.comparingInt(SPIOrder::getOrder));
     }
   }
   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

2. RegistrationManager.INSTANCE.run()

   public void run() {
       EventManager.getEventBus().register(new AfterServiceInstanceRegistryHandler(registrationList.size())); 
       registrationList.forEach(registration -> registration.run());
     }
   1
   2
   3
   4

3. Registration的具体实现-ServiceCenterRegistration，代理到RegistryUtils通过ServiceRegistry(RemoteServiceRegistry是其实现)和注册中心通信。

   • RemoteServiceRegistry，Rest请求，事件总线监听者模式，定时任务执行，这里可以后续细看细看，EventBus发挥着巨大的作用。
     • 定时执行ServiceCenterTask，包括各种Runnable的实现，定时执行参数从ServiceRegistryConfig中获取。
     • 定时心跳健康监测执行。
     • 通过ServiceRegistryClientImpl和注册中心Rest通信，注册服务到注册中心。在执行的逻辑中，会通过post各种事件，让外部监听者可以收到相应的请求。

4. Registration的具体实现-LocalRegistration，本地注册中心

DiscoveryManager-服务发现

Registration/Discovery-继承多接口将功能抽象分类可学

继承多个接口，每个接口都是高度抽象的功能，区分不同的接口，将方法分类，经典的面向对象的思路，解耦不同的方法，便于后续动态扩展。模板方法将面向接口的编程，使得当前的类功能非常容易扩展。

public interface Registration extends SPIEnabled, SPIOrder, LifeCycle
public interface Discovery extends SPIEnabled, SPIOrder, LifeCycle
1
2

• SPIEnabled，SPI加载进去是否使用。

• SPIOrder，SPI加载到内存中的排序。

• LifeCycle，当前类具有生命周期，模板方法相继调用不同的方法，如init、run，destory。

• ServiceCenterRegistration实现了Registration用于和注册中心交互，代理到RegistryUtils -> RemoteServiceRegistry和注册中心交互。

• ServiceCenterDiscovery实现了Discovery，同样代理到了RegistryUtils -> RemoteServiceRegistry和注册中心交互。

基于上述二位，我们有必要解读下RemoteServiceRegistry哈。

RemoteServiceRegistry-注意这个怎么和Eureka交互，底层交互有意思

这种类的设计方式和Apollo的类构造很像，分为本地仓库，远端仓库等模式，本地配置，远端配置等。

最终肯定会调用ServiceRegistryClient - > HttpClients 发起Rest请求。

重点关注如下配置？

1. Provider新增，多久会更新到Consumer服务本地？
2. Provider挂了，多久会更新到Consumer服务本地？
3. 这些参数是否支持可以配置？

结合下图理解如下方法流程：

• Provider心跳保活/注册，定时任务和注册中心保活provider，这是定时间隔serviceRegistryConfig.getHeartbeatInterval() /s，会定时执行ServiceCenterTask -> MicroserviceServiceCenterTask，包括如下Task，注册微服务监控，心跳，状态同步等。

  addTask(new MicroserviceRegisterTask(eventBus, srClient, microservice));
  addTask(new MicroserviceInstanceRegisterTask(eventBus, serviceRegistryConfig, srClient, microservice));
  addTask(new MicroserviceWatchTask(eventBus, serviceRegistryConfig, srClient, microservice));
  addTask(new MicroserviceInstanceHeartbeatTask(eventBus, srClient, microservice)); // 心跳心跳，活着么？
  addTask(new MicroserviceInstanceStatusSyncTask(eventBus, srClient, microservice));
  1
  2
  3
  4
  5

• Provider下线

• Consumer定时更新拉取Provider并更新本地服务缓存：定时间隔，serviceRegistryConfig.getInstancePullInterval() /s，如果服务失效需要移除服务。org.apache.servicecomb.serviceregistry.registry.cache.RefreshableServiceRegistryCache#refreshCache

寻找某个jar包由哪个pom文件引入

1、mvn dependency:tree -Dverbose -Dincludes=org.apache.servicecomb:java-chassis-core:2.1.5

2、mvn dependency:tree -Dverbose 找出全部依赖树，如果失败则执行向下install方法

@RpcSchema-Pojo类型调用的生产者

@RpcSchema(schemaId = "hello")
public class HelloImpl implements Hello {

  @Override
  public String sayHi(String name) {
    return "Hello " + name;
  }

  @Override
  public String sayHello(Person person) {
    return "Hello person " + person.getName();
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

• 注解会被PojoProducers这个BeanPostProcessor处理，用于往注册中心注册服务。

  protected void processProvider(String beanName, Object bean) {
      // 1、aop后，新的实例的父类可能是原class，也可能只是个proxy，父类不是原class，所以，需要先取出原class，再取标注
      // 调用 AopProxyUtils.ultimateTargetClass
      Class<?> beanCls = BeanUtils.getImplClassFromBean(bean);
      if (beanCls == null) {
        return;
      }
      RpcSchema rpcSchema = beanCls.getAnnotation(RpcSchema.class);
      if (rpcSchema == null) {
        return;
      }
  
      // 2、获取schemaId，如果没有传递则获取接口名
      String schemaId = rpcSchema.schemaId();
      if (StringUtils.isEmpty(schemaId)) {
        Class<?>[] intfs = beanCls.getInterfaces();
        if (intfs.length == 1) {
          schemaId = intfs[0].getName();
        } else {
          throw new Error("Must be schemaId or implements only one interface");
        }
      }
  
      // 3、注册producer元信息
      PojoProducerMeta pojoProducerMeta = new PojoProducerMeta();
      pojoProducerMeta.setSchemaId(schemaId); // 
      pojoProducerMeta.setSchemaInterface(rpcSchema.schemaInterface()); // 接口名
      pojoProducerMeta.setInstance(bean); // 实例类
  
      registerPojoProducer(pojoProducerMeta);
    }
  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

  • 在SCB启动的时候，scbEngine.getProducerProviderManager().getProducerProviderList(),获取全部ProducerProvider的实现类，用于获取Bean工厂中的生产提供者。包括从SPI中创建的，及使用Bean的形式注入的，默认主要有以下两种

    • PojoProducerProvider：scb run的时候，会调用PojoProducerProvider的Init方法从PojoProducers中获取全部的Provider元信息，注意在scbRun的时候，Bean工厂实例化全部的Bean实例。

      public List<ProducerMeta> init() {
          // for some test cases, there is no spring context
          if (BeanUtils.getContext() == null) {
            return Collections.emptyList();
          }
      
          PojoProducers pojoProducers = BeanUtils.getContext().getBean(PojoProducers.class);
          for (ProducerMeta producerMeta : pojoProducers.getProducerMetas()) {
            PojoProducerMeta pojoProducerMeta = (PojoProducerMeta) producerMeta;
            initPojoProducerMeta(pojoProducerMeta);
          }
      
          return pojoProducers.getProducerMetas();
        }
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14

    • RestProducerProvider：同理，从RestProducers中获取全部提供Rest接口的实例信息。

  • `会获取PojoProducerProvider -> 。

    
    1

• SCB.Run的时候，会通过获取到的元信息，将其注册到注册中心上对外提供服务。

@RestSchema-Rest形式调用的生产者

和RpcSchema处理方式类似，并且可以和RpcSchema的形式并存。

• 注解会被RestProducers处理，并将元信息保存。
• 在scb.run的时候，会被RestProducerProvider获取全部的元信息，并传递到SCB最终的控制中心中去。

@RpcReference-消费者

@RpcReference(microserviceName = "hello", schemaId = "hello")
private static Hello hello;
1
2

Pojo形式的Rpc消费方的注解，将自动从注册中心拉取对应的服务名，动态代理Compute并注入代理类。

• 注解会被RpcReferenceProcessor这个BeanPostProcessor处理，动态代理增强Hello。十分注意StringValueResolver支持动态解析占位符。

  public class RpcReferenceProcessor implements BeanPostProcessor, EmbeddedValueResolverAware {
    private StringValueResolver resolver;
   @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
      // 扫描所有field，处理扩展的field标注
      ReflectionUtils.doWithFields(bean.getClass(), new ReflectionUtils.FieldCallback() {
        public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException {
          processConsumerField(bean, field);
        }
      });
  
      return bean;
    }
    protected void processConsumerField(Object bean, Field field) {
      RpcReference reference = field.getAnnotation(RpcReference.class);
      if (reference == null) {
        return;
      }
  
      handleReferenceField(bean, field, reference);
    }
    private void handleReferenceField(Object obj, Field field,
        RpcReference reference) {
      String microserviceName = reference.microserviceName();
      microserviceName = resolver.resolveStringValue(microserviceName); // 微服务的名字支持占位符
      PojoReferenceMeta pojoReference = new PojoReferenceMeta();
      pojoReference.setMicroserviceName(microserviceName);
      pojoReference.setSchemaId(reference.schemaId());
      pojoReference.setConsumerIntf(field.getType());
  
  	// proxy = Invoker.createProxy(microserviceName, schemaId, consumerIntf); 动态代理
      pojoReference.afterPropertiesSet();
  
      ReflectionUtils.makeAccessible(field);
      ReflectionUtils.setField(field, obj, pojoReference.getProxy());
    } 
   }
  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

  1. 设置Consumer的元信息，包括服务的名字，SchemaId，契约接口名，最后根据接口名，生成动态代理类。
  2. 反射的方式注入field。必须makeAccessible使其支持private注入。

SPIServiceUtils-自定义SPI加载器

ServiceComb大量使用SPI动态扩展对应的实现，SPI定义的使能及优先级，从全部jar包中

META-INF/services的路径文件中加载扩展类，文件名为接口类的全路径里面内容为接口实现。

• SPIOrder，支持加载SPI顺序，排序从小到大，order数值越小，优先级越高，调用越靠前。

• SPIEnabled，支持SPI实现类是否使能。

• SPIServiceUtils，代理到JDK自带的ServiceLoader动态加载SPI类，并使用ReflectionUtils用于获取SPIOrder排序对应的实现类。

  // 加载SPI全部类
  public static <T> List<T> loadSortedService(Class<T> serviceType) {
      List<Entry<Integer, T>> serviceEntries = new ArrayList<>();
      ServiceLoader<T> serviceLoader = ServiceLoader.load(serviceType); // 代理到JDK
      serviceLoader.forEach(service -> {
        int serviceOrder = 0;
        // 实现getOrder方法就获取，否则就使用默认值，不强迫用户实现Order方法  
        Method getOrder = ReflectionUtils.findMethod(service.getClass(), "getOrder"); 
        if (getOrder != null) {
          serviceOrder = (int) ReflectionUtils.invokeMethod(getOrder, service); // 触发方法
        }
  
        Entry<Integer, T> entry = new SimpleEntry<>(serviceOrder, service);
        serviceEntries.add(entry);
      });
  
      List<T> services = serviceEntries.stream()
          .sorted(Comparator.comparingInt(Entry::getKey))
          .map(Entry::getValue)
          .collect(Collectors.toList()); // 根据 getOrder 排序，并返回全部的serviceType。
  
      LOGGER.info("Found SPI service {}, count={}.", serviceType.getName(), services.size());
      for (int idx = 0; idx < services.size(); idx++) {
        T service = services.get(idx);
        LOGGER.info("  {}. {}.", idx, service.getClass().getName());
      } // 输出加载的日志及其顺序，方便定位调试功能
  
      return services;
    }
  // Map, List