ServiceComb之Java

ServiceComb之Java"/>

ServiceComb之Java

项目网址：

pojo-provider项目的启动类是PojoProviderMain

public class PojoProviderMain {public static void main(String[] args) throws Exception {Log4jUtils.init();BeanUtils.init();}
}

如上所示，Log4jUtils.init是初始化日志服务。init方法代码如下，默认会从classpath*:config/base/log4j.properties和classpath*:config/log4j.properties两个路径读取log4j配置并合并。

public static void init() throws Exception {init(Arrays.asList("classpath*:config/base/log4j.properties", "classpath*:config/log4j.properties"));
}

init()方法的实现也就是读取配置文件，并调用PropertConfigurator进行初始化。

public static void init(List<String> locationPatterns) throws Exception {if (inited) {return;}synchronized (LOCK) {if (inited) {return;}PropertiesLoader loader = new PropertiesLoader(locationPatterns);Properties properties = loader.load();if (properties.isEmpty()) {throw new Exception("can not find resource " + locationPatterns);}PropertyConfigurator.configure(properties);inited = true;if (OUTPUT_CONFIG_ENABLED_TRUE.equals(properties.getProperty(OUTPUT_CONFIG_ENABLED, OUTPUT_CONFIG_ENABLED_TRUE))) {// If the property file with the highest priority is on a hard disk(not in a jar package)// and we have write access, output the merged property file for the purpose of debuggingoutputFile(loader.getFoundResList(), properties);}}
}

BeanUtils.init对微服务进行初始化，init方法代码如下

public static final String DEFAULT_BEAN_CORE_RESOURCE = "classpath*:META-INF/spring/scb-core-bean.xml";public static final String DEFAULT_BEAN_NORMAL_RESOURCE = "classpath*:META-INF/spring/*.bean.xml";public static final String[] DEFAULT_BEAN_RESOURCE = new String[] {DEFAULT_BEAN_CORE_RESOURCE, DEFAULT_BEAN_NORMAL_RESOURCE};public static void init() {init(DEFAULT_BEAN_RESOURCE);
}public static void init(String... configLocations) {prepareServiceCombScanPackage();Set<String> locationSet = new LinkedHashSet<>();addBeanLocation(locationSet, DEFAULT_BEAN_RESOURCE);addBeanLocation(locationSet, configLocations);context = new ClassPathXmlApplicationContext(locationSet.toArray(new String[locationSet.size()]));
}

prepareServiceCombScanPackage配置扫描的包的范围。

public static void prepareServiceCombScanPackage() {Set<String> scanPackags = new LinkedHashSet<>();// add exists settingsString exists = System.getProperty(SCB_SCAN_PACKAGE);if (exists != null) {for (String exist : exists.trim().split(",")) {if (!exist.isEmpty()) {addItem(scanPackags, exist.trim());}}}// ensure servicecomb package existaddItem(scanPackags, SCB_PACKAGE);// add main class packagefor (Class<?> mainClass : new Class<?>[] {JvmUtils.findMainClass(), JvmUtils.findMainClassByStackTrace()}) {if (mainClass != null && mainClass.getPackage() != null) {String pkg = mainClass.getPackage().getName();addItem(scanPackags, pkg);}}// finishString scbScanPackages = StringUtils.join(scanPackags, ",");System.setProperty(SCB_SCAN_PACKAGE, scbScanPackages);LOGGER.info("Scb scan package list: " + scbScanPackages);}

默认会从classpath*:META-INF/spring/*.bean.xml路径加载配置文件，并将其传递给Spring框架的ClassPathXmlApplicationContext完成应用上下文加载。在Spring context加载完成后，通过Spring Framework开源的事件调用机制和反射机制来触发ServiceComb自己的初始化逻辑。类似于观察者模型。首先我们知道Spring框架的AbstractApplicationContext中的refresh函数是用来加载和刷新spring配置。

/*** Create a new ClassPathXmlApplicationContext, loading the definitions* from the given XML files and automatically refreshing the context.* @param configLocations array of resource locations* @throws BeansException if context creation failed*/
public ClassPathXmlApplicationContext(String... configLocations) throws BeansException {this(configLocations, true, null);
}

public ClassPathXmlApplicationContext(String[] configLocations, boolean refresh, ApplicationContext parent)throws BeansException {super(parent);setConfigLocations(configLocations);if (refresh) {refresh();}
}

其中，refresh方法完成了整个spring生命周期的初始化，参考博客。

@Override
public void refresh() throws BeansException, IllegalStateException {synchronized (this.startupShutdownMonitor) {// 准备，记录容器的启动时间startupDate, 标记容器为激活，初始化上下文环境如文件路径信息，验证必填属性是否填写prepareRefresh();// 获取新的beanFactory，销毁原有beanFactory、为每个bean生成BeanDefinition等ConfigurableListableBeanFactory beanFactory = obtainFreshBeanFactory();// 初始化beanfactory的各种属性prepareBeanFactory(beanFactory);try {// 模板方法，此时，所有的beanDefinition已经加载，但是还没有实例化。//允许在子类中对beanFactory进行扩展处理。比如添加aware相关接口自动装配设置，添加后置处理器等，是子类扩展prepareBeanFactory(beanFactory)的方法postProcessBeanFactory(beanFactory);// 实例化并调用所有注册的beanFactory后置处理器（实现接口BeanFactoryPostProcessor的bean，在beanFactory标准初始化之后执行）invokeBeanFactoryPostProcessors(beanFactory);// 实例化和注册beanFactory中扩展了BeanPostProcessor的bean//例如：// AutowiredAnnotationBeanPostProcessor(处理被@Autowired注解修饰的bean并注入)// RequiredAnnotationBeanPostProcessor(处理被@Required注解修饰的方法)// CommonAnnotationBeanPostProcessor(处理@PreDestroy、@PostConstruct、@Resource等多个注解的作用)等。registerBeanPostProcessors(beanFactory);// Initialize message source for this context.initMessageSource();// Initialize event multicaster for this context.initApplicationEventMulticaster();// 模板方法，在容器刷新的时候可以自定义逻辑，不同的Spring容器做不同的事情。onRefresh();// 注册监听器，广播early application eventsregisterListeners();// 实例化所有剩余的（非懒加载）单例// 比如invokeBeanFactoryPostProcessors方法中根据各种注解解析出来的类，在这个时候都会被初始化。// 实例化的过程各种BeanPostProcessor开始起作用。finishBeanFactoryInitialization(beanFactory);// refresh做完之后需要做的其他事情。// 清除上下文资源缓存（如扫描中的ASM元数据）// 初始化上下文的生命周期处理器，并刷新（找出Spring容器中实现了Lifecycle接口的bean并执行start()方法）。// 发布ContextRefreshedEvent事件告知对应的ApplicationListener进行响应的操作finishRefresh();} catch (BeansException ex) {if (logger.isWarnEnabled()) {logger.warn("Exception encountered during context initialization - " +"cancelling refresh attempt: " + ex);}// Destroy already created singletons to avoid dangling resources.destroyBeans();// Reset 'active' flag.cancelRefresh(ex);// Propagate exception to caller.throw ex;} finally {// Reset common introspection caches in Spring's core, since we// might not ever need metadata for singleton beans anymore...resetCommonCaches();}}
}

spring容器初始化过程如下所示：

在finishRefresh方法中通过publishEvent循环通知所有的观察者（ApplicationEvent的子类）的onApplicationEvent方法执行相应的操作，通过它我们可以定位到CseApplicationListener。

protected void finishRefresh() {// Initialize lifecycle processor for this context.initLifecycleProcessor();// Propagate refresh to lifecycle processor first.getLifecycleProcessor().onRefresh();// Publish the final event.publishEvent(new ContextRefreshedEvent(this));// Participate in LiveBeansView MBean, if active.LiveBeansView.registerApplicationContext(this);
}

/*** Publish the given event to all listeners.* @param event the event to publish (may be an {@link ApplicationEvent}* or a payload object to be turned into a {@link PayloadApplicationEvent})* @param eventType the resolved event type, if known* @since 4.2*/
protected void publishEvent(Object event, @Nullable ResolvableType eventType) {Assert.notNull(event, "Event must not be null");// Decorate event as an ApplicationEvent if necessaryApplicationEvent applicationEvent;if (event instanceof ApplicationEvent) {applicationEvent = (ApplicationEvent) event;}else {applicationEvent = new PayloadApplicationEvent<>(this, event);if (eventType == null) {eventType = ((PayloadApplicationEvent<?>) applicationEvent).getResolvableType();}}// Multicast right now if possible - or lazily once the multicaster is initializedif (this.earlyApplicationEvents != null) {this.earlyApplicationEvents.add(applicationEvent);}else {getApplicationEventMulticaster().multicastEvent(applicationEvent, eventType);}// Publish event via parent context as well...if (this.parent != null) {if (this.parent instanceof AbstractApplicationContext) {((AbstractApplicationContext) this.parent).publishEvent(event, eventType);}else {this.parent.publishEvent(event);}}
}

@FunctionalInterface
public interface ApplicationListener<E extends ApplicationEvent> extends EventListener {/*** Handle an application event.* @param event the event to respond to*/void onApplicationEvent(E event);}

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.INSTANCE.init();DiscoveryManager.INSTANCE.init();}public void setInitEventClass(Class<?> initEventClass) {this.initEventClass = initEventClass;}@Overridepublic int getOrder() {// should run before default listener, eg: ZuulConfigurationreturn -1000;}@Overridepublic void onApplicationEvent(ApplicationEvent event) {if (initEventClass.isInstance(event)) {if (applicationContext instanceof AbstractApplicationContext) {((AbstractApplicationContext) applicationContext).registerShutdownHook();}SCBEngine scbEngine = SCBEngine.getInstance();//SCBEngine init first, hence we do not need worry that when other beans need use the//producer microserviceMeta, the SCBEngine is not inited.
//        String serviceName = RegistryUtils.getMicroservice().getServiceName();
//        SCBEngine.getInstance().setProducerMicroserviceMeta(new MicroserviceMeta(serviceName).setConsumer(false));
//        SCBEngine.getInstance().setProducerProviderManager(applicationContext.getBean(ProducerProviderManager.class));
//        SCBEngine.getInstance().setConsumerProviderManager(applicationContext.getBean(ConsumerProviderManager.class));
//        SCBEngine.getInstance().setTransportManager(applicationContext.getBean(TransportManager.class));scbEngine.setApplicationContext(applicationContext);scbEngine.setFilterChainsManager(applicationContext.getBean(FilterChainsManager.class));scbEngine.getConsumerProviderManager().getConsumerProviderList().addAll(applicationContext.getBeansOfType(ConsumerProvider.class).values());scbEngine.getProducerProviderManager().getProducerProviderList().addAll(applicationContext.getBeansOfType(ProducerProvider.class).values());scbEngine.addBootListeners(applicationContext.getBeansOfType(BootListener.class).values());scbEngine.run();} else if (event instanceof ContextClosedEvent) {if (SCBEngine.getInstance() != null) {SCBEngine.getInstance().destroy();}}}
}

scbEngine.run()方法如下：

  public synchronized SCBEngine run() {if (SCBStatus.DOWN.equals(status)) {try {doRun();waitStatusUp();} catch (TimeoutException e) {LOGGER.warn("{}", e.getMessage());} catch (Throwable e) {LOGGER.error("Failed to start ServiceComb due to errors and close", e);try {destroy();} catch (Exception exception) {LOGGER.info("destroy has some error.", exception);}status = SCBStatus.FAILED;throw new IllegalStateException("ServiceComb init failed.", e);} finally {printServiceInfo();}}return this;}

doRun()方法如下：

private void doRun() throws Exception {status = SCBStatus.STARTING;bootListeners.sort(ComparatorparingInt(BootListener::getOrder));triggerEvent(EventType.BEFORE_HANDLER);HandlerConfigUtils.init(consumerHandlerManager, producerHandlerManager);triggerEvent(EventType.AFTER_HANDLER);triggerEvent(EventType.BEFORE_FILTER);filterChainsManager.init();triggerEvent(EventType.AFTER_FILTER);createProducerMicroserviceMeta();triggerEvent(EventType.BEFORE_PRODUCER_PROVIDER);producerProviderManager.init();triggerEvent(EventType.AFTER_PRODUCER_PROVIDER);triggerEvent(EventType.BEFORE_CONSUMER_PROVIDER);consumerProviderManager.init();triggerEvent(EventType.AFTER_CONSUMER_PROVIDER);triggerEvent(EventType.BEFORE_TRANSPORT);transportManager.init(this);triggerEvent(EventType.AFTER_TRANSPORT);triggerEvent(EventType.BEFORE_REGISTRY);triggerAfterRegistryEvent();RegistrationManager.INSTANCE.run();DiscoveryManager.INSTANCE.run();shutdownHook = new Thread(this::destroyForShutdownHook);Runtime.getRuntime().addShutdownHook(shutdownHook);}

具体CSE初始化过程如下所示：

• 首先HandlerConfigUtils.init读取配置文件信息cse.handler.xml （HandlerConfigUtils.init()）
• filterChainsManager.init()完成对应filter的处理；
• 然后producerProviderManager.init()进行Provider处理，主要是微服务元数据、Schema和契约的处理，并将其添加到RegistryUtils的微服务实例中；
• consumerProviderManager.init()完成服务消费者的初始化；
• 接着transportManager.init(this)启动一个异步无阻塞的网络框架vertx，并将初始化的EndPoint添加到RegistryUtils的微服务实例中；
• 最后RegistrationManager.INSTANCE.run()在RegistryUtils中实现微服务在服务管理中心的注册。
  
  在前面的流程分析中，CseApplicationListener类很关键，相当于Spring框架与ServiceComb的桥梁。在Spring初始化之后触发，初始化ServiceComb自身的配置。

初始化过程分析

CseApplicationListener最终触发SCBEngine.doInit方法，triggerEvent是一种AOP机制。
进一步，在HandlerConfigUtils中，可以看到处理链handler实例的初始化。首先获取了handler处理链的配置信息，然后根据这些配置初始化所有的handler实例。

public static void init(ConsumerHandlerManager consumerHandlerManager, ProducerHandlerManager producerHandlerManager)throws Exception {Config config = loadConfig();consumerHandlerManager.init(config);producerHandlerManager.init(config);
}

那么它会从哪里获取配置呢？

源码如下，可以看到其会获取类路径下config/cse.handler.xml的文件。其实断点进去PaaSResourceUtils.getSortedResources方法可知还会获取类路径下config/cse.*.handler.xml文件。在ProducerProviderManager初始化中，我们可以得到微服务的元数据信息microserviceMeta，并把它添加到RegistryUtils中定义的微服务当中↓

private static Config loadConfig() throws Exception {Config config = new Config();List<Resource> resList =PaaSResourceUtils.getSortedResources("classpath*:config/cse.handler.xml", ".handler.xml");for (Resource res : resList) {Config tmpConfig = XmlLoaderUtils.load(res, Config.class);config.mergeFrom(tmpConfig);}return config;
}

createProducerMicroserviceMeta()创建服务元数据，通过断点可知，微服务的元数据主要包括如下的内容：

TransportManager.init主要包括如下内容。transport.init()会启动一个异步无阻塞的网络框架vertx。启动成功后将初始化的EndPoint添加到RegistryUtils的微服务实例中。

  public void init(SCBEngine scbEngine) throws Exception {buildTransportMap();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;}}}