Spring Cloud Stream(一):Spring基础知识

 我研究和阅读Spring Cloud Stream源码已经有一个多月了,但是由于自己的Spring基础知识不是很充足,所以导致很多地方都没有融会贯通,并且相关的文章一直无从下手。于是我先整理了当时阅读代码时的知识点记录,算是源码分析之前的基础知识储备吧,整理的有些杂乱,希望大家理解。
 本文涉及的Spring知识如下:

  • Spring Boot的@Import用法和原理,与ConfigurationImportBeanDefinitionRegistrar相关
  • Bean初始化各个周期的回调,比如InitializingBean,BeanPostProcessor,SmartInitializingSingleton
  • FactoryBeanMethodInterceptor
  • Aware系列回调
  • LifecycleSmartLifecycleDefaultLifecycleProcessor

BeanDefinitionRegistryPostProcessor

BeanDefinitionRegistryPostProcessor实现了BeanFactoryPostProcessor接口,是Spring框架的BeanDefinitionRegistry的后处理器,用来注册额外的BeanDefinitionpostProcessBeanDefinitionRegistry方法会在所有的BeanDefinition已经被加载了,但是所有的Bean还没有被创建前调用。BeanDefinitionRegistryPostProcessor经常被用来注册BeanFactoryPostProcessorBeanDefinition

ImportBeanDefinitionRegistrar

@Import注解用来支持在Configuration类中引入其他的配置类,包括Configuration类,ImportSelectorImportBeanDefinitionRegistrar的实现类。ImportBeanDefinitionRegistrarConfigurationClassPostProcessor处理Configuration类期间被调用,用来生成该Configuration类所需要的BeanDefinition。而ConfigurationClassPostProcessor正实现了BeanDefinitionRegistryPostProcessor接口。下面我们就来看一下其processConfigBeanDefinitions方法到底是如何处理Configuration类的。

public void processConfigBeanDefinitions(BeanDefinitionRegistry registry) {
		List<BeanDefinitionHolder> configCandidates = new ArrayList<>();
		String[] candidateNames = registry.getBeanDefinitionNames();
        //第一步:先把所有Configuration的beanDefinition找到。
		for (String beanName : candidateNames) {
			BeanDefinition beanDef = registry.getBeanDefinition(beanName);
			//利用AnnotationMetadata是否有@Configuration这个注解。需要注意的是
            //Configuration是一个元注解,它是可以使用在其他注解上的,被这些注解注释的类也被认为是Configuration
            if (ConfigurationClassUtils.checkConfigurationClassCandidate(beanDef, this.metadataReaderFactory)) {
				configCandidates.add(new BeanDefinitionHolder(beanDef, beanName));
			}
		}
		//第二步:通过Order注解的值来排序,定义了Configuration的先后顺序
		configCandidates.sort((bd1, bd2) -> {
			int i1 = ConfigurationClassUtils.getOrder(bd1.getBeanDefinition());
			int i2 = ConfigurationClassUtils.getOrder(bd2.getBeanDefinition());
			return (i1 < i2) ? -1 : (i1 > i2) ? 1 : 0;
		});
        //..... 此处有省略
		ConfigurationClassParser parser = new ConfigurationClassParser(
				this.metadataReaderFactory, this.problemReporter, this.environment,
				this.resourceLoader, this.componentScanBeanNameGenerator, registry);

		Set<BeanDefinitionHolder> candidates = new LinkedHashSet<>(configCandidates);
		Set<ConfigurationClass> alreadyParsed = new HashSet<>(configCandidates.size());
		do {
            //第三步:通过BeanDefinition来读取ConfigurationClass
			parser.parse(candidates);
			parser.validate();

			Set<ConfigurationClass> configClasses = new LinkedHashSet<>(parser.getConfigurationClasses());
			configClasses.removeAll(alreadyParsed);

			if (this.reader == null) {
				this.reader = new ConfigurationClassBeanDefinitionReader(
						registry, this.sourceExtractor, this.resourceLoader, this.environment,
						this.importBeanNameGenerator, parser.getImportRegistry());
			}
            //第四步:重点,通过ConfigurationClass来获得BeanDefinition
			this.reader.loadBeanDefinitions(configClasses);
			alreadyParsed.addAll(configClasses);

			candidates.clear();
            //第五步:由于在loadBeanDefinitions过程中会向registry中添加BeanDefinition,所以这里需要把新的Definition
            //在重新检测一遍,先看是否是Configuration类,如果是的那么还要再进行一次处理。
			if (registry.getBeanDefinitionCount() > candidateNames.length) {
                //.....此处有省略,大致逻辑就是通过registry多出的BeanDefinition获得新的candidateNames
				candidateNames = newCandidateNames;
			}
		}
		while (!candidates.isEmpty());
        //.....此处有省略
	}

 接着我们直接到ConfigurationClassBeanDefinitionReader类中查看loadBeanDefinition函数的实现。它会调用loadBeanDefinitionsForConfigurationClass函数。在该函数中会处理所有和Configuration相关的BeanDefinition,其中就会调用loadBeanDefinitionsFromRegistrars来通过ImportBeanDefinitionRegistrar加载BeanDefinition
 看到这里,大家可能会有个疑问,多个Configuration和多个ImportBeanDefinitionRegistrar存在的情况下,它们之间的对应关系是如何确定的呢?
ConfigurationClassParser的parse方法会将Configuration类相关的配置信息全部解析出来。我们可以看其doProcessConfigurationClass方法的源码。通过@Import注解将Configuration类和相应的ImportBeanDefinitionRegistrar联系在一起。

protected final SourceClass doProcessConfigurationClass(ConfigurationClass configClass, SourceClass sourceClass)
			throws IOException {
        //首先处理内部成员类的情况
		processMemberClasses(configClass, sourceClass);

		// 处理 @PropertySource 注解
		for (AnnotationAttributes propertySource : AnnotationConfigUtils.attributesForRepeatable(
				sourceClass.getMetadata(), PropertySources.class,
				org.springframework.context.annotation.PropertySource.class)) {
			if (this.environment instanceof ConfigurableEnvironment) {
				processPropertySource(propertySource);
            }
		}

		// 处理 @ComponentScan 注解
		Set<AnnotationAttributes> componentScans = AnnotationConfigUtils.attributesForRepeatable(
				sourceClass.getMetadata(), ComponentScans.class, ComponentScan.class);
		if (!componentScans.isEmpty() &&
				!this.conditionEvaluator.shouldSkip(sourceClass.getMetadata(), ConfigurationPhase.REGISTER_BEAN)) {
			for (AnnotationAttributes componentScan : componentScans) {
				// The config class is annotated with @ComponentScan -> perform the scan immediately
				Set<BeanDefinitionHolder> scannedBeanDefinitions =
						this.componentScanParser.parse(componentScan, sourceClass.getMetadata().getClassName());
				// Check the set of scanned definitions for any further config classes and parse recursively if needed
				for (BeanDefinitionHolder holder : scannedBeanDefinitions) {
					if (ConfigurationClassUtils.checkConfigurationClassCandidate(
							holder.getBeanDefinition(), this.metadataReaderFactory)) {
						parse(holder.getBeanDefinition().getBeanClassName(), holder.getBeanName());
					}
				}
			}
		}

		// 处理 @Import 注解
		processImports(configClass, sourceClass, getImports(sourceClass), true);

		// 处理 @ImportResource 注解
		if (sourceClass.getMetadata().isAnnotated(ImportResource.class.getName())) {
			AnnotationAttributes importResource =
					AnnotationConfigUtils.attributesFor(sourceClass.getMetadata(), ImportResource.class);
			String[] resources = importResource.getStringArray("locations");
			Class<? extends BeanDefinitionReader> readerClass = importResource.getClass("reader");
			for (String resource : resources) {
				String resolvedResource = this.environment.resolveRequiredPlaceholders(resource);
				configClass.addImportedResource(resolvedResource, readerClass);
			}
		}

		// 处理configuration中的 @Bean 函数
		Set<MethodMetadata> beanMethods = retrieveBeanMethodMetadata(sourceClass);
		for (MethodMetadata methodMetadata : beanMethods) {
			configClass.addBeanMethod(new BeanMethod(methodMetadata, configClass));
		}
        //......有省略
		return null;
	}

InitializingBean,FactoryBean,MethodInterceptor

 Spring Cloud Stream的BindableProxyFactory类实现了上述接口。

BindableProxyFactory implements MethodInterceptor, FactoryBean<Object>, Bindable, InitializingBean

 其中,InitializingBean接口有一个afterPropertiesSet方法,该方法在bean所有的属性都被赋值后调用。bean的属性被初始化是在初始化的时候做的,与BeanPostProcessor结合来看,afterPropertiesSet方法在postProcessBeforeInitializationpostProcessAfterInitialization之间被调用。
 Spring中有两个类型的Bean,普通Bean和工厂Bean。FactoryBean有三个接口,分别是:

  • Object getObject():返回FactoryBean创建的对象实例。
  • boolean isSingleton():表示FactoryBean返回的对象实例是否为单例。
  • Class getObjectType():返回FactoryBean返回的对象类型。
     我们可以看一下BindableProxyFactory的相关实现,这里会和MethodInterceptor配合。MethodInterceptor是AOP相关的接口,用于在调用对象接口时进行切片注入或在直接实现接口。
    
    @Override
    	public synchronized Object getObject() throws Exception {
            //使用AOP的ProxyFactory类,由于该类本身也是先了MethodInterceptor接口
            //所以这样配合使用,直接返回ProxyFactory类。
    		if (this.proxy == null) {
    			ProxyFactory factory = new ProxyFactory(this.type, this);
    			this.proxy = factory.getProxy();
    		}
    		return this.proxy;
    	}
    
    	@Override
    	public Class<?> getObjectType() {
    		return this.type;
    	}
    
    	@Override
    	public boolean isSingleton() {
    		return true;
    	}
    

BeanPostProcessor,ApplicationContextAware,BeanFactoryAware,SmartInitializingSingleton

 Spring Cloud Stream的StreamListenerAnnotationBeanPostProcessor实现了如下接口

public class StreamListenerAnnotationBeanPostProcessor
		implements BeanPostProcessor, ApplicationContextAware, BeanFactoryAware, SmartInitializingSingleton,
		InitializingBean

BeanPostProcessorbean的后处理器,通过它我们可以在Bean初始化前后进行处理。它的postProcessBeforeInitialization方法在Bean初始化之前被调用,而postProcessAfterInitializationBean初始化后被调用。相关原理涉及到Spring创建Bean的流程,这个之后有时间再研究吧。

Aware系列接口

 Spring中提供了一些Aware相关的接口,像是BeanFactoryAware,ApplicationContextAware等。当一个类实现了这些接口之后,Aware接口的Bean在初始化之后,可以取得相应的资源的实例。比如StreamListenerAnnotationBeanPostProcessor对象就实现了ApplicationContextAwareBeanFactoryAware接口来获取ConfigurableApplicationContextBeanFactory实例。

SmartInitializingSingleton

 当所有的singleton的bean都初始化完成之后才会调用这个接口
afterSingletonsInstantiated函数

SmartLifecycle

 之前介绍的接口都是在Bean的生命周期内的某个阶段中被调用,如果我们希望在容器本身的生命周期事件上做一些事情该怎麽办呢?Spring容器提供了Lifecycle接口。当ApplicationContext接口启动或在关闭时,它会调用本容器内所有的Lifecycle接口。

public interface Lifecycle {
    //启动该组件
	void start();
    //停止组件
	void stop();
    //查看组件是否正在运行
	boolean isRunning();

}

 如果两个对象有依赖关系,希望某一个bean先初始化完成,完成一些工作之后,再初始化另一个bean。在这个场景下,可以使用SmartLifecycle接口,该接口的getPhase方法返回一个整型数字,表明执行顺序。如果其getPhase()方法返回Integer.MIN_VALUE,那么该对象最先启动,最后停止;如果返回Integer.MAX_VALUE,那么该对象最后启动,最先停止。在Spring容器里,有DefaultLifecycleProcessor这个类来处理所有的Lifecycle的bean。在AbstractApplicationContextfinishRefresh函数中会调用到该processer的onRefresh函数,从其调用其本身的startBeans函数。

private void startBeans(boolean autoStartupOnly) {
		Map<String, Lifecycle> lifecycleBeans = getLifecycleBeans();
		Map<Integer, LifecycleGroup> phases = new HashMap<Integer, LifecycleGroup>();
        //遍历所有的Lifecycle,按照phase分成不同的LifecycleGroup
		for (Map.Entry<String, ? extends Lifecycle> entry : lifecycleBeans.entrySet()) {
			Lifecycle bean = entry.getValue();
			if (!autoStartupOnly || (bean instanceof SmartLifecycle && ((SmartLifecycle) bean).isAutoStartup())) {
				int phase = getPhase(bean);
				LifecycleGroup group = phases.get(phase);
				if (group == null) {
					group = new LifecycleGroup(phase, this.timeoutPerShutdownPhase, lifecycleBeans, autoStartupOnly);
					phases.put(phase, group);
				}
				group.add(entry.getKey(), bean);
			}
		}
		if (!phases.isEmpty()) {
            //按照phase排序,然后启动
			List<Integer> keys = new ArrayList<Integer>(phases.keySet());
			Collections.sort(keys);
			for (Integer key : keys) {
				phases.get(key).start();
			}
		}
	}

1000 Compartir