CSE RPC流程分析

1 使用方法

1.1 Provider

使用示例如下：

@RpcSchema(schemaId = "hello")
//@RpcSchema
public class HelloImpl implements Hello {@Overridepublic String sayHi(String name) {return "Hello " + name;}@Overridepublic String sayHello(Person person) {return "Hello person " + person.getName();}
}

1.2 Consumer

使用示例如下：

@Component
public class CustomHandlerCustomerMain {// 表示使用的provider的信息@RpcReference(microserviceName = "customerhandler", schemaId = "hello")private static Hello hello;public static void main(String[] args) throws InterruptedException {BeanUtils.init();System.out.println(hello.sayHi("Java Chassis"));Person person = new Person();person.setName("ServiceComb/Java Chassis");System.out.println(hello.sayHello(person));sleep(1000);}
}

2 信息初始化

cse和spring的启动过程此处省略。

2.1 Consumer初始化过程

RpcReferenceProcessor类扫描RpcReference注解：

@Component
public class RpcReferenceProcessor implements BeanPostProcessor, EmbeddedValueResolverAware {private StringValueResolver resolver;@Overridepublic Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {// 1 扫描所有field，处理扩展的field标注，递归扫描所有声明变量ReflectionUtils.doWithFields(bean.getClass(), new ReflectionUtils.FieldCallback() {public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException {processConsumerField(bean, field);}});return bean;}@Overridepublic Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {return bean;}protected void processConsumerField(Object bean, Field field) {RpcReference reference = field.getAnnotation(RpcReference.class);if (reference == null) {return;}// 2 添加了RpcReference注解的变量都会进入该方法handleReferenceField(bean, field, reference);}@Overridepublic void setEmbeddedValueResolver(StringValueResolver resolver) {this.resolver = resolver;}private void handleReferenceField(Object obj, Field field,RpcReference reference) {String microserviceName = reference.microserviceName();// 3 解析microserviceName，可以使用通配符microserviceName = resolver.resolveStringValue(microserviceName);PojoReferenceMeta pojoReference = new PojoReferenceMeta();pojoReference.setMicroserviceName(microserviceName);pojoReference.setSchemaId(reference.schemaId());// 4 所有interface接口信息，创建代理使用pojoReference.setConsumerIntf(field.getType());// 5 创建动态代理对象pojoReference.afterPropertiesSet();ReflectionUtils.makeAccessible(field);// 6 设置对象为动态代理对象ReflectionUtils.setField(field, obj, pojoReference.getProxy());}
}

其中，第5步，pojoReference.afterPropertiesSet()，创建动态代理，PojoReferenceMeta类的实现如下：

public class PojoReferenceMeta implements FactoryBean<Object>, InitializingBean {// 原始数据private String microserviceName;private String schemaId;// consumerIntf意思是consumer interfaceprivate Class<?> consumerIntf;// 根据intf创建出来的动态代理// TODO:未实现本地优先(本地场景下，应该跳过handler机制)private Object proxy;......@Overridepublic void afterPropertiesSet() {if (consumerIntf == null) {throw new ServiceCombException(String.format("microserviceName=%s, schemaid=%s, \n"+ "do not support implicit interface anymore, \n"+ "because that caused problems:\n"+ "  1.the startup process relies on other microservices\n"+ "  2.cyclic dependent microservices can not be deployed\n"+ "suggest to use @RpcReference or "+ "<cse:rpc-reference id=\"...\" microservice-name=\"...\" schema-id=\"...\" interface=\"...\"></cse:rpc-reference>.",microserviceName,schemaId));}proxy = Invoker.createProxy(microserviceName, schemaId, consumerIntf);}
}

其中，创建proxy过程在Invoker类中实现，Invoker类如下：

public class Invoker implements InvocationHandler {protected final PojoConsumerMetaRefresher metaRefresher;protected final PojoInvocationCreator invocationCreator;protected final DefaultMethodMeta defaultMethodMeta = new DefaultMethodMeta();protected InvocationCaller invocationCaller;// 创建动态代理的实现@SuppressWarnings("unchecked")public static <T> T createProxy(String microserviceName, String schemaId, Class<?> consumerIntf) {Invoker invoker = new Invoker(microserviceName, schemaId, consumerIntf);// 创建代理，调用方法时，会调用其中的public Object invoke(Object proxy, Method method, Object[] args)方法// Proxy的实现为java.lang.reflect.Proxyreturn (T) Proxy.newProxyInstance(consumerIntf.getClassLoader(), new Class<?>[] {consumerIntf}, invoker);}public Invoker(String microserviceName, String schemaId, Class<?> consumerIntf) {this.metaRefresher = createInvokerMeta(microserviceName, schemaId, consumerIntf);this.invocationCreator = createInvocationCreator();}protected PojoConsumerMetaRefresher createInvokerMeta(String microserviceName, String schemaId,Class<?> consumerIntf) {return new PojoConsumerMetaRefresher(microserviceName, schemaId, consumerIntf);}public PojoInvocationCreator createInvocationCreator() {return new PojoInvocationCreator();}protected InvocationCaller createInvocationCaller() {if (SCBEngine.getInstance().isFilterChainEnabled()) {return new FilterInvocationCaller();}return new HandlerInvocationCaller();}@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {if (method.isDefault()) {return defaultMethodMeta.getOrCreateMethodHandle(proxy, method).invokeWithArguments(args);}SCBEngine.getInstance().ensureStatusUp();prepareInvocationCaller();return invocationCaller.call(method, metaRefresher, invocationCreator, args);}protected void prepareInvocationCaller() {if (invocationCaller != null) {return;}this.invocationCaller = createInvocationCaller();}
}

后面，还会进行更加深入的介绍，初始化过程到此结束。

2.2 Provider初始化过程

在CSE初始化过程中，会调用如下过程：

public class SCBEngine {// 类的构造方法protected SCBEngine() {......producerProviderManager = new ProducerProviderManager(this);serviceRegistryListener = new ServiceRegistryListener(this);}// 初始化过程中调用方法private void doRun() throws Exception {createProducerMicroserviceMeta();triggerEvent(EventType.BEFORE_PRODUCER_PROVIDER);producerProviderManager.init();triggerEvent(EventType.AFTER_PRODUCER_PROVIDER);......RegistrationManager.INSTANCE.run();}
}

ProducerProviderManager类的部分实现过程如下：

public class ProducerProviderManager {private List<ProducerProvider> producerProviderList = new ArrayList<>(SPIServiceUtils.getOrLoadSortedService(ProducerProvider.class));public void init() {registerProducerMetas(producerMetas);for (ProducerProvider provider : producerProviderList) {List<ProducerMeta> producerMetas = provider.init();if (producerMetas == null) {LOGGER.warn("ProducerProvider {} not provide any producer.", provider.getClass().getName());continue;}registerProducerMetas(producerMetas);}}
}

producerProviderList通过SPI方式初始化，SPI的配置信息如下：

org.apache.servicecomb.provider.pojo.PojoProducerProvider

在ProducerProviderManager的init方法中，对producerProviderList进行逐一init初始化和处理。

RpcSchema注解使用PojoProducerProvider进行扫描，PojoProducerProvider的init方法如下：

@Override
public List<ProducerMeta> init() {// for some test cases, there is no spring contextif (BeanUtils.getContext() == null) {return Collections.emptyList();}PojoProducers pojoProducers = BeanUtils.getContext().getBean(PojoProducers.class);for (ProducerMeta producerMeta : pojoProducers.getProducerMetas()) {PojoProducerMeta pojoProducerMeta = (PojoProducerMeta) producerMeta;// cse 提供了多种provider，pojoprovider实现方式不需要单独初始化initPojoProducerMeta(pojoProducerMeta);}return pojoProducers.getProducerMetas();
}

PojoProducers扫描RpcSchema注解，并将接口契约信息保存入类的producerMetas变量中，PojoProducers的实现过程如下：

public class PojoProducers implements BeanPostProcessor {private List<ProducerMeta> producerMetas = new ArrayList<>();public synchronized void registerPojoProducer(PojoProducerMeta pojoProducer) {producerMetas.add(pojoProducer);}......@Overridepublic Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {processProvider(beanName, bean);return bean;}protected void processProvider(String beanName, Object bean) {// aop后，新的实例的父类可能是原class，也可能只是个proxy，父类不是原class// 所以，需要先取出原class，再取标注Class<?> beanCls = BeanUtils.getImplClassFromBean(bean);if (beanCls == null) {return;}RpcSchema rpcSchema = beanCls.getAnnotation(RpcSchema.class);if (rpcSchema == null) {return;}// 如果未设置schemaId，则会使用路径名称作为schemaIdString 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");}}PojoProducerMeta pojoProducerMeta = new PojoProducerMeta();pojoProducerMeta.setSchemaId(schemaId);pojoProducerMeta.setSchemaInterface(rpcSchema.schemaInterface());pojoProducerMeta.setInstance(bean);registerPojoProducer(pojoProducerMeta);}
}

PojoProducerProvider的init方法执行完成后，ProducerProviderManager的registerProducerMetas方法会执行，实现如下：

private void registerProducerMetas(List<ProducerMeta> producerMetas) {for (ProducerMeta producerMeta : producerMetas) {registerSchema(producerMeta.getSchemaId(), producerMeta.getSchemaInterface(), producerMeta.getInstance());}
}public SchemaMeta registerSchema(String schemaId, Class<?> schemaInterface, Object instance) {MicroserviceMeta producerMicroserviceMeta = scbEngine.getProducerMicroserviceMeta();Swagger swagger = scbEngine.getSwaggerLoader().loadLocalSwagger(producerMicroserviceMeta.getAppId(),producerMicroserviceMeta.getShortName(),schemaId);SwaggerProducer swaggerProducer = scbEngine.getSwaggerEnvironment().createProducer(instance, schemaInterface, swagger);swagger = swaggerProducer.getSwagger();registerUrlPrefixToSwagger(swagger);SchemaMeta schemaMeta = producerMicroserviceMeta.registerSchemaMeta(schemaId, swagger);schemaMeta.putExtData(CoreMetaUtils.SWAGGER_PRODUCER, swaggerProducer);Executor reactiveExecutor = scbEngine.getExecutorManager().findExecutorById(ExecutorManager.EXECUTOR_REACTIVE);for (SwaggerProducerOperation producerOperation : swaggerProducer.getAllOperations()) {OperationMeta operationMeta = schemaMeta.ensureFindOperation(producerOperation.getOperationId());operationMeta.setSwaggerProducerOperation(producerOperation);if (CompletableFuture.class.equals(producerOperation.getProducerMethod().getReturnType())) {operationMeta.setExecutor(scbEngine.getExecutorManager().findExecutor(operationMeta, reactiveExecutor));}}return schemaMeta;}

执行完registerProducerMetas后，producer的信息就保存在producerMetas中。

在SCBEngine中执行完初始化信息后，会执行RegistrationManager.INSTANCE.run()方法：

public class RegistrationManager {// value is ip or {interface name}public static final String PUBLISH_ADDRESS = "servicecomb.service.publishAddress";private static final String PUBLISH_PORT = "servicecomb.{transport_name}.publishPort";private final List<Registration> registrationList = new ArrayList<>();private Registration primary;// 1 构造函数，初始化registrationList，Registration实现有两个：ServiceCenterRegistration和LocalRegistration，在cse中，本地注册中心和远程注册中心是两个不同jar包，可以通过pom文件进行选择private RegistrationManager() {SPIServiceUtils.getOrLoadSortedService(Registration.class).stream().filter((SPIEnabled::enabled)).forEach(registrationList::add);initPrimary();}public void run() {// 创建EventBus消息EventManager.getEventBus().register(new AfterServiceInstanceRegistryHandler(registrationList.size()));// 此处执行ServiceCenterRegistration的run方法（以此为例）registrationList.forEach(registration -> registration.run());}public void init() {BeanUtils.addBeans(Registration.class, registrationList);initPrimary();registrationList.forEach(registration -> registration.init());}private void initPrimary() {if (registrationList.isEmpty()) {LOGGER.warn("No registration is enabled. Fix this if only in unit tests.");primary = null;} else {primary = registrationList.get(0);}}
}

此外，RegistrationManager的init方法在CseApplicationListener类中已经进行了初始化：

public class CseApplicationListenerimplements ApplicationListener<ApplicationEvent>, Ordered, ApplicationContextAware {private Class<?> initEventClass = ContextRefreshedEvent.class;private ApplicationContext applicationContext;@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {if (this.applicationContext == applicationContext) {// same object. avoid initialize many times.return;}this.applicationContext = applicationContext;BeanUtils.setContext(applicationContext);HttpClients.load();// 此处进行RegistrationManager初始化RegistrationManager.INSTANCE.init();DiscoveryManager.INSTANCE.init();}public void setInitEventClass(Class<?> initEventClass) {this.initEventClass = initEventClass;}}

ServiceCenterRegistration类实现如下：

public class ServiceCenterRegistration implements Registration {public static final String NAME = "service center registration";@Overridepublic void init() {RegistryUtils.init();}@Overridepublic void run() {RegistryUtils.run();}@Overridepublic void destroy() {RegistryUtils.destroy();}@Overridepublic int getOrder() {return Const.SERVICE_CENTER_ORDER;}@Overridepublic String name() {return NAME;}@Overridepublic MicroserviceInstance getMicroserviceInstance() {return RegistryUtils.getMicroserviceInstance();}@Overridepublic Microservice getMicroservice() {return RegistryUtils.getMicroservice();}@Overridepublic String getAppId() {return RegistryUtils.getAppId();}@Overridepublic boolean updateMicroserviceInstanceStatus(MicroserviceInstanceStatus status) {RegistryUtils.executeOnEachServiceRegistry(sr -> new SuppressedRunnableWrapper(() -> {MicroserviceInstance selfInstance = sr.getMicroserviceInstance();sr.getServiceRegistryClient().updateMicroserviceInstanceStatus(selfInstance.getServiceId(),selfInstance.getInstanceId(),status);}).run());return true;}@Overridepublic void addSchema(String schemaId, String content) {RegistryUtils.executeOnEachServiceRegistry(sr -> {sr.getMicroservice().addSchema(schemaId, content);});}@Overridepublic void addEndpoint(String endpoint) {RegistryUtils.executeOnEachServiceRegistry(sr -> {Microservice microservice = sr.getMicroservice();microservice.getInstance().getEndpoints().add(endpoint);});}@Overridepublic void addBasePath(Collection<BasePath> basePaths) {RegistryUtils.executeOnEachServiceRegistry(sr -> {sr.getMicroservice().getPaths().addAll(basePaths);});}@Overridepublic boolean enabled() {return DynamicPropertyFactory.getInstance().getBooleanProperty(Const.SERVICE_CENTER_ENABLED, true).get();}
}

调用的均为RegistryUtils中的方法，具体实现如下：

import com.google.common.eventbus.Subscribe;public final class RegistryUtils {private static final Logger LOGGER = LoggerFactory.getLogger(RegistryUtils.class);/*** The default ServiceRegistry instance*/private static volatile ServiceRegistry serviceRegistry;private static final Map<String, ServiceRegistry> EXTRA_SERVICE_REGISTRIES = new LinkedHashMap<>();private static AggregateServiceRegistryCache aggregateServiceRegistryCache;private RegistryUtils() {}public static synchronized void init() {if (serviceRegistry != null) {return;}// 1 ConfigUtil.createLocalConfig()会加载microservice.yml的配置，包括cse内部和业务自定义// 2 调用initializeServiceRegistriesWithConfig方法initializeServiceRegistriesWithConfig(ConfigUtil.createLocalConfig());initAggregateServiceRegistryCache();}private static void initAggregateServiceRegistryCache() {ArrayList<ServiceRegistry> serviceRegistries = new ArrayList<>();executeOnEachServiceRegistry(serviceRegistries::add);aggregateServiceRegistryCache = new AggregateServiceRegistryCache(serviceRegistries);aggregateServiceRegistryCache.setCacheRefreshedWatcher(refreshedCaches -> DiscoveryManager.INSTANCE.getAppManager().pullInstances());executeOnEachServiceRegistry(serviceRegistry -> serviceRegistry.getEventBus().register(aggregateServiceRegistryCache));}private static void initializeServiceRegistriesWithConfig(Configuration configuration) {// 3 serviceRegistry初始化为RemoteServiceRegistry类serviceRegistry =ServiceRegistryFactory.create(ServiceRegistryConfig.INSTANCE, configuration);// 4 调用initializeServiceRegistries方法进行initializeServiceRegistries(configuration);}private static void initializeServiceRegistries(Configuration configuration) {Map<String, ServiceRegistryConfig> configs = BeanUtils.getBeansOfType(ServiceRegistryConfig.class);configs.forEach((k, v) -> {ServiceRegistry serviceRegistry = ServiceRegistryFactory.create(v, configuration);addExtraServiceRegistry(serviceRegistry);});// 4.1 此处调用RemoteServiceRegistry的init方法，在init方法中会创建对应的MicroserviceRegisterTask、MicroserviceInstanceRegisterTask、MicroserviceWatchTask、MicroserviceInstanceHeartbeatTask、MicroserviceInstanceStatusSyncTask、ServiceCenterTask，以及建立本地的信息缓存executeOnEachServiceRegistry(ServiceRegistry::init);// 4.2 使用guava的EventBus事件总线，监听MicroserviceInstanceRegisterTask处理事件，进行微服务信息注册完成后的处理executeOnEachServiceRegistry(AfterServiceInstanceRegistryHandler::new);}public static void run() {// 5 执行MicroserviceRegisterTask、MicroserviceInstanceRegisterTask、MicroserviceWatchTask、MicroserviceInstanceHeartbeatTask、MicroserviceInstanceStatusSyncTask、ServiceCenterTaskexecuteOnEachServiceRegistry(ServiceRegistry::run);}
}

初始化的ServiceRegistryConfig过程为：

public ServiceRegistryConfig build() {return new ServiceRegistryConfig().setHttpVersion(getHttpVersion()).setInstances(getInstances()).setIpPort(getIpPort()).setSsl(isSsl()).setClientName(RegistryHttpClientOptionsSPI.CLIENT_NAME).setWatchClientName(RegistryWatchHttpClientOptionsSPI.CLIENT_NAME).setConnectionTimeout(getConnectionTimeout()).setIdleConnectionTimeout(getIdleConnectionTimeout()).setIdleWatchConnectionTimeout(getIdleWatchTimeout()).setRequestTimeout(getRequestTimeout()).setHeartBeatRequestTimeout(getHeartBeatRequestTimeout()).setHeartbeatInterval(getHeartbeatInterval()).setInstancePullInterval(getInstancePullInterval()).setRegistryAutoDiscovery(isRegistryAutoDiscovery()).setResendHeartBeatTimes(getResendHeartBeatTimes()).setAlwaysOverrideSchema(isAlwaysOverrideSchema()).setIgnoreSwaggerDifference(isIgnoreSwaggerDifference()).setPreferIpAddress(isPreferIpAddress()).setWatch(isWatch()).setRegistryApiVersion(getRegistryApiVersion()).setTenantName(getTenantName()).setDomainName(getDomainName()).setAccessKey(getAccessKey()).setSecretKey(getSecretKey()).setProxyEnable(isProxyEnable()).setProxyHost(getProxyHost()).setProxyPort(getProxyPort()).setProxyUsername(getProxyUsername()).setProxyPasswd(getProxyPasswd()).setAuthHeaderProviders(getAuthHeaderProviders());
}

初始化过程基本介绍完成，继续回到RegistrationManager的run方法中，会执行registrationList.forEach(registration -> registration.run())方法，此时MicroserviceRegisterTask、MicroserviceInstanceRegisterTask、MicroserviceWatchTask、MicroserviceInstanceHeartbeatTask、MicroserviceInstanceStatusSyncTask、ServiceCenterTask就会执行起来。

回到consumer的初始化过程，在SCBEngine中DiscoveryManager.INSTANCE.run()方法会进行微服务信息的发现操作。

同时，可以看到cse使用EventBus进行事件通信，spring使用listener进行事件通信，EventBus比较简单，listener功能更加强大，两者都是基于观察者设计模式。

编码：RegistrationManager-》ServiceCenterRegistration-》RegistryUtils层层递进，结构清晰，SCBEngine中做统一的管理。

3 调用过程

1 Consumer

使用的测试方法如下：

@Component
public class CustomHandlerCustomerMain {@RpcReference(microserviceName = "customerhandler", schemaId = "hello")private static Hello hello;public static void main(String[] args) {BeanUtils.init();System.out.println(hello.sayHi("Java Chassis"));Person person = new Person();person.setName("ServiceComb/Java Chassis");System.out.println(hello.sayHello(person));}
}

1、首先执行hello.sayHi(“Java Chassis”)方法，会进入代理类的invoke方法中

package org.apache.servicecomb.provider.pojo;public class Invoker implements InvocationHandler {protected final PojoConsumerMetaRefresher metaRefresher;protected final PojoInvocationCreator invocationCreator;protected final DefaultMethodMeta defaultMethodMeta = new DefaultMethodMeta();protected InvocationCaller invocationCaller;@Overridepublic Object invoke(Object proxy, Method method, Object[] args) throws Throwable {// 1 如果是接口的default实现，直接创建default方法的handleif (method.isDefault()) {return defaultMethodMeta.getOrCreateMethodHandle(proxy, method).invokeWithArguments(args);}// cse包含五种状态：//DOWN：Chassis is Down// STARTING：Chassis is Starting (progressing)// UP：Chassis is Running// STOPPING：Chassis is Stopping (progressing)// FAILED：Chassis Init Failed// 2 只有当cse处于UP状态时，请求才能发出去SCBEngine.getInstance().ensureStatusUp();// 3 准备处理过程函数prepareInvocationCaller();// 5 调用方法return invocationCaller.call(method, metaRefresher, invocationCreator, args);}protected void prepareInvocationCaller() {if (invocationCaller != null) {return;}this.invocationCaller = createInvocationCaller();}protected InvocationCaller createInvocationCaller() {// 4 cse 提供filter调用处理链和handler调用处理链，filter处理链默认为falseif (SCBEngine.getInstance().isFilterChainEnabled()) {return new FilterInvocationCaller();}return new HandlerInvocationCaller();}
}

在上面第4步中，由于cse core模块中定义了如下microservice.yaml，因此默认不使用filter-chains。

servicecomb:exception:invocation:print-stack-trace: falsefilter-chains:enabled: falsetransport:scb-consumer-transport:rest: rest-client-codec, rest-client-senderscb-producer-transport:rest: rest-server-codecdefinition:scb-consumer: simple-load-balance, scb-consumer-transportscb-producer: scb-producer-transport, schedule, producer-operationconsumer:framework: scb-consumerproducer:framework: scb-producer

第4步中，创建HandlerInvocationCaller，其实现如下：

package org.apache.servicecomb.provider.pojo;public class HandlerInvocationCaller implements InvocationCaller {@Overridepublic Object call(Method method, PojoConsumerMetaRefresher metaRefresher, PojoInvocationCreator invocationCreator,Object[] args) {// 5.1 创建本次调用的上下文信息PojoInvocation invocation = invocationCreator.create(method, metaRefresher, args);// 5.2 本次调用中使用同步调用方法if (invocation.isSync()) {return syncInvoke(invocation);}return completableFutureInvoke(invocation);}protected Object syncInvoke(PojoInvocation invocation) {// 5.3 进行方法远程调用Response response = InvokerUtils.innerSyncInvoke(invocation);if (response.isSucceed()) {// 5.4 方法调用成功，将object转换为对应的类return invocation.convertResponse(response);}throw ExceptionFactory.convertConsumerException(response.getResult());}protected CompletableFuture<Object> completableFutureInvoke(PojoInvocation invocation) {CompletableFuture<Object> future = new InvocationContextCompletableFuture<>(invocation);InvokerUtils.reactiveInvoke(invocation, response -> {if (response.isSucceed()) {Object result = invocation.convertResponse(response);future.complete(result);return;}future.completeExceptionally(response.getResult());});return future;}
}

5.1中invocationCaller的实现如下：

public class PojoInvocationCreator {// PojoConsumerMetaRefresher为consumer信息刷新方法，用来更新从registercenter获取的信息public PojoInvocation create(Method method, PojoConsumerMetaRefresher metaRefresher, Object[] args) {PojoConsumerMeta pojoConsumerMeta = metaRefresher.getLatestMeta();PojoConsumerOperationMeta consumerOperationMeta = pojoConsumerMeta.ensureFindOperationMeta(method);PojoInvocation invocation = new PojoInvocation(consumerOperationMeta);invocation.setSuccessResponseType(consumerOperationMeta.getResponsesType());invocation.setInvocationArguments(consumerOperationMeta.getSwaggerConsumerOperation().toInvocationArguments(args));invocation.setSync(consumerOperationMeta.isSync());return invocation;}
}

5.3中innerSyncInvoke实现如下：

/*** This is an internal API, caller make sure already invoked SCBEngine.ensureStatusUp* @param invocation* @return servicecomb response object*/
public static Response innerSyncInvoke(Invocation invocation) {try {// 5.3.1 enableEventLoopBlockingCallCheck默认为trueif (enableEventLoopBlockingCallCheck && isInEventLoop()) {throw new IllegalStateException("Can not execute sync logic in event loop. ");}invocation.onStart(null, System.nanoTime());// 5.3.2 创建执行器并进行初始化SyncResponseExecutor respExecutor = new SyncResponseExecutor();invocation.setResponseExecutor(respExecutor);// onStartHandlersRequest实现参加下面，初始化请求开始处理时间invocation.onStartHandlersRequest();// 5.3.3 【重要】执行业务自定义和默认的handler调用链invocation.next(respExecutor::setResponse);// 5.3.4 等待方法执行，具体实现逻辑参加下面SyncResponseExecutorResponse response = respExecutor.waitResponse(invocation);// 5.3.5 记录本次调用结束时间invocation.getInvocationStageTrace().finishHandlersResponse();// 5.3.6 发送执行成功事件invocation.onFinish(response);return response;} catch (Throwable e) {String msg =String.format("invoke failed, %s", invocation.getOperationMeta().getMicroserviceQualifiedName());LOGGER.error(msg, e);Response response = Response.createConsumerFail(e);invocation.onFinish(response);return response;}
}// 其中，isInEventLoop的实现如下，cse底层使用vertx框架作为通信框架
// Vert.x是一个基于JVM、轻量级、高性能的应用平台，非常适用于最新的移动端后台、互联网、企业应用架构。Vert.x基于全异步Java服务器Netty，并扩展出了很多有用的特性。(Netty是一个高性能、异步事件驱动的NIO框架，它提供了对TCP、UDP和文件传输的支持，作为一个异步NIO框架)netty比较关心网络传输这个层面的东西，重写了nio里面的buffer，channel，和一些基本的序列化实现。面向应用的需要还需要用户自己来做一些轻量的封装。所以很多知名的框架都是基于netty实现，akka,vert.x等等。static boolean isOnEventLoopThread() {Thread t = Thread.currentThread();return t instanceof VertxThread && !((VertxThread)t).isWorker();}/*** 业务线程在阻塞等待着，不必另起线程* 将应答流程包装为Runnable，先唤醒业务线程，再在业务线程中执行runnable*/
public class SyncResponseExecutor implements Executor {private CountDownLatch latch;private Runnable cmd;private Response response;public SyncResponseExecutor() {latch = new CountDownLatch(1);}@Overridepublic void execute(Runnable cmd) {this.cmd = cmd;// one network thread, many connections, then this notify will be performance bottlenecks// if save to a queue, and other thread(s) to invoke countDown, will get good performance// but if have multiple network thread, this "optimization" will reduce performance// now not change this.latch.countDown();}// 5.3.4 waitResponse方法public Response waitResponse(Invocation invocation) throws InvocationException {// 5.3.4.1 waitResponse方法guardedWait(invocation);// cmd为null，是没走execute，直接返回的场景if (cmd != null) {cmd.run();}return response;}public void setResponse(Response response) {this.response = response;if (cmd == null) {// 1. 走到这里，没有cmd，说明没走到网络线程，直接就返回了。// 2. 或者在网络线程中没使用execute的方式返回，这会导致返回流程在网络线程中执行，虽然不合适，但是也不应该导致业务线程无法唤醒latch.countDown();}}private void guardedWait(Invocation invocation) throws InvocationException {// // 5.3.4.2 获取等待时间long wait = getWaitTime(invocation);try {// 等待时间小于0，一直等待if (wait <= 0) {latch.await();return;}// 5.3.4.3 等待execute方法执行latch.countDown()，返回true，调用可以执行if (latch.await(wait, TimeUnit.MILLISECONDS)) {return;}} catch (InterruptedException e) {//ignore}throw new InvocationException(REQUEST_TIMEOUT, ExceptionCodes.INVOCATION_TIMEOUT, "Invocation Timeout.");}private long getWaitTime(Invocation invocation) {if (invocation.getOperationMeta().getConfig().getMsRequestTimeout() <= 0) {return invocation.getOperationMeta().getConfig().getMsInvocationTimeout();}if (invocation.getOperationMeta().getConfig().getMsInvocationTimeout() <= 0) {return invocation.getOperationMeta().getConfig().getMsRequestTimeout();}return Math.min(invocation.getOperationMeta().getConfig().getMsRequestTimeout(),invocation.getOperationMeta().getConfig().getMsInvocationTimeout());}
}

下面对5.3.3进行详细解释：

invocation的部分实现如下，handlerList初始化后，最后一个handler一定是TransportClientHandler，一定会执行TransportClientHandler的handle方法：

package org.apache.servicecomb.core;import com.fasterxml.jackson.databind.JavaType;public class Invocation extends SwaggerInvocation {public void next(AsyncResponse asyncResp) throws Exception {// 不必判断有效性，因为整个流程都是内部控制的int runIndex = handlerIndex;handlerIndex++;// handlerList初始化后，最后一个handler一定是TransportClientHandlerhandlerList.get(runIndex).handle(this, asyncResp);}
}

TransportClientHandler实现如下，在handler中执行transport.send方法：

public class TransportClientHandler implements Handler {private static final Logger log = LoggerFactory.getLogger(TransportClientHandler.class);public static final TransportClientHandler INSTANCE = new TransportClientHandler();@Overridepublic void handle(Invocation invocation, AsyncResponse asyncResp) throws Exception {Transport transport = invocation.getTransport();log.debug("Sending request {} to {}",invocation.getMicroserviceQualifiedName(),invocation.getEndpoint().getEndpoint());transport.send(invocation, asyncResp);}
}

transport的send方法实现如下：

public class VertxRestTransport extends AbstractTransport {@Overridepublic void send(Invocation invocation, AsyncResponse asyncResp) throws Exception {restClient.send(invocation, asyncResp);}
}

restClient的send方法实现如下：

public class RestTransportClient {private static final Logger LOGGER = LoggerFactory.getLogger(RestTransportClient.class);private List<HttpClientFilter> httpClientFilters;public void init(Vertx vertx) throws Exception {httpClientFilters = SPIServiceUtils.getSortedService(HttpClientFilter.class);}public void send(Invocation invocation, AsyncResponse asyncResp) {HttpClientWithContext httpClientWithContext = findHttpClientPool(invocation);RestClientInvocation restClientInvocation = new RestClientInvocation(httpClientWithContext, httpClientFilters);try {restClientInvocation.invoke(invocation, asyncResp);} catch (Throwable e) {asyncResp.fail(invocation.getInvocationType(), e);LOGGER.error("vertx rest transport send error.", e);}}
}

restClientInvocation的invoke实现如下：

public class RestClientInvocation {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();// 从注册中心获取的契约信息中获取请求服务器的ip地址和portFuture<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);}// 新建VertxClientRequestToHttpServletRequest请求HttpServletRequestEx requestEx = new VertxClientRequestToHttpServletRequest(clientRequest, requestBodyBuffer);invocation.getInvocationStageTrace().startClientFiltersRequest();// 发送请求前的filter处理，ClientRestArgsFilter，RestTemplateCopyHeaderFilter，DefaultHttpClientFilterfor (HttpClientFilter filter : httpClientFilters) {if (filter.enabled()) {filter.beforeSendRequest(invocation, requestEx);}}// 从业务线程转移到网络线程中去发送invocation.onStartSendRequest();httpClientWithContext.runOnContext(httpClient -> {clientRequest.setTimeout(operationMeta.getConfig().getMsRequestTimeout());clientRequest.response().onComplete(asyncResult -> {if (asyncResult.failed()) {fail(asyncResult.cause());return;}handleResponse(asyncResult.result());});processServiceCombHeaders(invocation, operationMeta);restClientRequest.end().onComplete((t) -> invocation.getInvocationStageTrace().finishWriteToBuffer(System.nanoTime()));});return Future.succeededFuture();}).onFailure(failure -> {invocation.getTraceIdLogger().error(LOGGER, "Failed to send request, alreadyFailed:{}, local:{}, remote:{}, message={}.",alreadyFailed, getLocalAddress(), ipPort.getSocketAddress(),ExceptionUtils.getExceptionMessageWithoutTrace(failure));throwableHandler.handle(failure);});}
}

最终rpc调用通过vertx实现的http请求发送给provider执行。

2 Provider

provider对外提供服务，入口为VertxRestDispatcher的onRequest方法：

package org.apache.servicecomb.transport.rest.vertx;public class VertxRestDispatcher extends AbstractVertxHttpDispatcher {private static final Logger LOGGER = LoggerFactory.getLogger(VertxRestDispatcher.class);private static final String KEY_ORDER = "servicecomb.http.dispatcher.rest.order";private static final String KEY_ENABLED = "servicecomb.http.dispatcher.rest.enabled";private static final String KEY_PATTERN = "servicecomb.http.dispatcher.rest.pattern";private Transport transport;private MicroserviceMeta microserviceMeta;public VertxRestDispatcher() {}public void init(Router router) {String pattern = DynamicPropertyFactory.getInstance().getStringProperty("servicecomb.http.dispatcher.rest.pattern", (String)null).get();if (pattern == null) {router.route().handler(this.createBodyHandler());router.route().failureHandler(this::failureHandler).handler(this::onRequest);} else {router.routeWithRegex(pattern).handler(this.createBodyHandler());router.routeWithRegex(pattern).failureHandler(this::failureHandler).handler(this::onRequest);}}protected void onRequest(RoutingContext context) {if (transport == null) {transport = SCBEngine.getInstance().getTransportManager().findTransport(Const.RESTFUL);microserviceMeta = SCBEngine.getInstance().getProducerMicroserviceMeta();}// 请求体初始化HttpServletRequestEx requestEx = new VertxServerRequestToHttpServletRequest(context);HttpServletResponseEx responseEx = new VertxServerResponseToHttpServletResponse(context.response());// 默认不执行if (SCBEngine.getInstance().isFilterChainEnabled()) {InvocationCreator creator = new RestVertxProducerInvocationCreator(context,microserviceMeta, transport.getEndpoint(),requestEx, responseEx);new RestProducerInvocationFlow(creator, requestEx, responseEx).run();return;}// 默认执行下面的代码VertxRestInvocation vertxRestInvocation = new VertxRestInvocation();context.put(RestConst.REST_PRODUCER_INVOCATION, vertxRestInvocation);// 执行方法vertxRestInvocation.invoke(transport, requestEx, responseEx, httpServerFilters);}
}

VertxRestInvocation的实现如下：

public class VertxRestInvocation extends RestProducerInvocation {@Overrideprotected void createInvocation() {super.createInvocation();RoutingContext routingContext = ((VertxServerRequestToHttpServletRequest) this.requestEx).getContext();VertxHttpTransportContext transportContext = new VertxHttpTransportContext(routingContext, requestEx, responseEx,produceProcessor);invocation.setTransportContext(transportContext);routingContext.put(RestConst.REST_INVOCATION_CONTEXT, this.invocation);}
}public class RestProducerInvocation extends AbstractRestInvocation {protected Transport transport;public void invoke(Transport transport, HttpServletRequestEx requestEx, HttpServletResponseEx responseEx,List<HttpServerFilter> httpServerFilters) {this.transport = transport;this.requestEx = requestEx;this.responseEx = responseEx;this.httpServerFilters = httpServerFilters;requestEx.setAttribute(RestConst.REST_REQUEST, requestEx);try {// 1 找到对应的执行信息findRestOperation();} catch (InvocationException e) {sendFailResponse(e);return;}// 2 执行对应方法scheduleInvocation();}protected void findRestOperation() {// 1.1 找到provider的契约信息MicroserviceMeta selfMicroserviceMeta = SCBEngine.getInstance().getProducerMicroserviceMeta();findRestOperation(selfMicroserviceMeta);}@Overrideprotected OperationLocator locateOperation(ServicePathManager servicePathManager) {// 1.2.1 根据uri，找到对应的method信息return servicePathManager.producerLocateOperation(requestEx.getRequestURI(), requestEx.getMethod());}@Overrideprotected void createInvocation() {this.invocation = InvocationFactory.forProvider(transport.getEndpoint(),restOperationMeta.getOperationMeta(),null);}
}public abstract class AbstractRestInvocation {protected void findRestOperation(MicroserviceMeta microserviceMeta) {ServicePathManager servicePathManager = ServicePathManager.getServicePathManager(microserviceMeta);if (servicePathManager == null) {LOGGER.error("No schema defined for {}:{}.", microserviceMeta.getAppId(), microserviceMeta.getMicroserviceName());throw new InvocationException(Status.NOT_FOUND, Status.NOT_FOUND.getReasonPhrase());}// 1.2 找到对应的method信息OperationLocator locator = locateOperation(servicePathManager);requestEx.setAttribute(RestConst.PATH_PARAMETERS, locator.getPathVarMap());this.restOperationMeta = locator.getOperation();}protected void scheduleInvocation() {try {// 2.1 设置vertx相关TransportContext信息，同时校验cse是否处于UP状态，否则不对外提供服务createInvocation();} catch (Throwable e) {sendFailResponse(e);return;}try {// 2.1 上下文信息设置，包括traceId和调用方microservice namethis.setContext();} catch (Exception e) {LOGGER.error("failed to set invocation context", e);sendFailResponse(e);return;}// 2.2 traceId设置，处理起始时间设置invocation.onStart(requestEx, start);invocation.getInvocationStageTrace().startSchedule();// 2.3 获取调用的provider的方法的信息OperationMeta operationMeta = restOperationMeta.getOperationMeta();// 流控相关，此处未使用Holder<Boolean> qpsFlowControlReject = checkQpsFlowControl(operationMeta);if (qpsFlowControlReject.value) {return;}try {operationMeta.getExecutor().execute(() -> {synchronized (this.requestEx) {try {if (isInQueueTimeout()) {throw new InvocationException(Status.INTERNAL_SERVER_ERROR, "Timeout when processing the request.");}if (requestEx.getAttribute(RestConst.REST_REQUEST) != requestEx) {// already timeout// in this time, request maybe recycled and reused by web container, do not use requestExLOGGER.error("Rest request already timeout, abandon execute, method {}, operation {}.",operationMeta.getHttpMethod(),operationMeta.getMicroserviceQualifiedName());return;}// 2.4 执行方法runOnExecutor();} catch (InvocationException e) {LOGGER.error("Invocation failed, cause={}", e.getMessage());sendFailResponse(e);} catch (Throwable e) {LOGGER.error("Processing rest server request error", e);sendFailResponse(e);}}});} catch (Throwable e) {LOGGER.error("failed to schedule invocation, message={}, executor={}.", e.getMessage(), e.getClass().getName());sendFailResponse(e);}}protected void runOnExecutor() {// 2.4.1 开始执行时间记录invocation.onExecuteStart();// 2.4.2 调用方法invoke();}public void invoke() {try {// 2.4.2.1 主要是执行HttpServerFilter的实现类Response response = prepareInvoke();// 2.4.2.2 如果有HttpServerFilter的实现类返回异常，直接返回if (response != null) {sendResponseQuietly(response);return;}// 2.4.2.3 执行方法调用doInvoke();} catch (InvocationException e) {LOGGER.error("Invocation failed, cause={}", e.getMessage());sendFailResponse(e);} catch (Throwable e) {LOGGER.error("Processing rest server request error", e);sendFailResponse(e);}}protected Response prepareInvoke() throws Throwable {this.initProduceProcessor();invocation.getHandlerContext().put(RestConst.REST_REQUEST, requestEx);invocation.getInvocationStageTrace().startServerFiltersRequest();for (HttpServerFilter filter : httpServerFilters) {if (filter.enabled()) {Response response = filter.afterReceiveRequest(invocation, requestEx);if (response != null) {return response;}}}return null;}protected void doInvoke() throws Throwable {invocation.onStartHandlersRequest();// 2.4.2.3 此时会进入invocation类中invocation.next(resp -> sendResponseQuietly(resp));}
}public class Invocation extends SwaggerInvocation {public void next(AsyncResponse asyncResp) throws Exception {// 不必判断有效性，因为整个流程都是内部控制的int runIndex = handlerIndex;handlerIndex++;// 2.4.2.3 handlerList初始化时，最后一个handler默认是ProducerOperationHandler，此处会执行进入ProducerOperationHandler类中的handle方法中handlerList.get(runIndex).handle(this, asyncResp);}
}public class ProducerOperationHandler implements Handler {private static final Logger LOGGER = LoggerFactory.getLogger(ProducerOperationHandler.class);public static final ProducerOperationHandler INSTANCE = new ProducerOperationHandler();@Overridepublic void handle(Invocation invocation, AsyncResponse asyncResp) {SwaggerProducerOperation producerOperation = invocation.getOperationMeta().getSwaggerProducerOperation();if (producerOperation == null) {asyncResp.producerFail(ExceptionUtils.producerOperationNotExist(invocation.getSchemaId(),invocation.getOperationName()));return;}// 2.4.2.3.1 执行方法invoke(invocation, producerOperation, asyncResp);}private void invoke(Invocation invocation, SwaggerProducerOperation producerOperation, AsyncResponse asyncResp) {if (CompletableFuture.class.equals(producerOperation.getProducerMethod().getReturnType())) {completableFutureInvoke(invocation, producerOperation, asyncResp);return;}// 2.4.2.3.2 同步调用syncInvoke(invocation, producerOperation, asyncResp);}public void syncInvoke(Invocation invocation, SwaggerProducerOperation producerOperation, AsyncResponse asyncResp) {ContextUtils.setInvocationContext(invocation);// 2.4.2.3.3 调用方法Response response = doInvoke(invocation, producerOperation);ContextUtils.removeInvocationContext();asyncResp.handle(response);}public Response doInvoke(Invocation invocation, SwaggerProducerOperation producerOperation) {Response response;try {// 2.4.2.3.4 调用业务方法时间记录invocation.onBusinessMethodStart();Object[] args = invocation.toProducerArguments();// ProducerInvokeExtension可通过引入cse相关jar包进行参数校验for (ProducerInvokeExtension producerInvokeExtension : producerOperation.getProducerInvokeExtenstionList()) {producerInvokeExtension.beforeMethodInvoke(invocation, producerOperation, args);}// 2.4.2.3.5 执行业务方法，此时通过反射真正进入业务方法Object result = producerOperation.getProducerMethod().invoke(producerOperation.getProducerInstance(), args);response = producerOperation.getResponseMapper().mapResponse(invocation.getStatus(), result);invocation.onBusinessMethodFinish();invocation.onBusinessFinish();} catch (Throwable e) {if (shouldPrintErrorLog(e)) {invocation.getTraceIdLogger().error(LOGGER, "unexpected error operation={}, message={}",invocation.getInvocationQualifiedName(),org.apache.servicecomb.foundation.common.utils.ExceptionUtils.getExceptionMessageWithoutTrace(e));}invocation.onBusinessMethodFinish();invocation.onBusinessFinish();response = processException(invocation, e);}return response;}protected boolean shouldPrintErrorLog(Throwable throwable) {if (!(throwable instanceof InvocationTargetException)) {return true;}Throwable targetException = ((InvocationTargetException) throwable).getTargetException();return !(targetException instanceof InvocationException);}protected Response processException(SwaggerInvocation invocation, Throwable e) {if (e instanceof InvocationTargetException) {e = ((InvocationTargetException) e).getTargetException();}return ExceptionFactory.convertExceptionToResponse(invocation, e);}
}// 业务实现如下
@RpcSchema(schemaId = "hello")
//@RpcSchema
public class HelloImpl implements Hello {@Overridepublic String sayHi(String name) {return "Hello " + name;}@Overridepublic String sayHello(Person person) {return "Hello person " + person.getName();}
}