Spring EnableJms 注解

源码分析

  1. @Target(ElementType.TYPE)
  2. @Retention(RetentionPolicy.RUNTIME)
  3. @Documented
  4. @Import(JmsBootstrapConfiguration.class)
  5. public @interface EnableJms {
  6. }
  • 该类的切入点在@Import(JmsBootstrapConfiguration.class) , 直接看JmsBootstrapConfiguration就可以了
  1. @Configuration
  2. @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  3. public class JmsBootstrapConfiguration {
  5.     /**
  6.      * jms 监听注解后处理, 将{@link JmsListener} 注册到{@link JmsListenerContainerFactory}
  7.      * @return
  8.      */
  9.     @Bean(name = JmsListenerConfigUtils.JMS_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
  10.     @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  11.     public JmsListenerAnnotationBeanPostProcessor jmsListenerAnnotationProcessor() {
  12.         return new JmsListenerAnnotationBeanPostProcessor();
  13.     }
  16.     /**
  17.      * JMS 监听注册
  18.      * @return
  19.      */
  20.     @Bean(name = JmsListenerConfigUtils.JMS_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
  21.     public JmsListenerEndpointRegistry defaultJmsListenerEndpointRegistry() {
  22.         return new JmsListenerEndpointRegistry();
  23.     }
  25. }

JmsListenerAnnotationBeanPostProcessor

类图

image-20200304085303580

  • 主要关注

    1. afterSingletonsInstantiated
  1. postProcessAfterInitialization

afterSingletonsInstantiated

  1. @Override
  2.     public void afterSingletonsInstantiated() {
  3.         // Remove resolved singleton classes from cache
  4.         this.nonAnnotatedClasses.clear();
  6.         if (this.beanFactory instanceof ListableBeanFactory) {
  7.             // Apply JmsListenerConfigurer beans from the BeanFactory, if any
  8.             // 根据类型获取bean
  9.             Map<String, JmsListenerConfigurer> beans =
  10.                     ((ListableBeanFactory) this.beanFactory).getBeansOfType(JmsListenerConfigurer.class);
  12.             List<JmsListenerConfigurer> configurers = new ArrayList<>(beans.values());
  13.             // 排序 Order
  14.             AnnotationAwareOrderComparator.sort(configurers);
  15.             for (JmsListenerConfigurer configurer : configurers) {
  16.                 // 放入jms监听配置,开发者自定义
  17.                 configurer.configureJmsListeners(this.registrar);
  18.             }
  19.         }
  21.         if (this.containerFactoryBeanName != null) {
  22.             this.registrar.setContainerFactoryBeanName(this.containerFactoryBeanName);
  23.         }
  25.         if (this.registrar.getEndpointRegistry() == null) {
  26.             // Determine JmsListenerEndpointRegistry bean from the BeanFactory
  27.             if (this.endpointRegistry == null) {
  28.                 Assert.state(this.beanFactory != null, "BeanFactory must be set to find endpoint registry by bean name");
  29.                 this.endpointRegistry = this.beanFactory.getBean(
  30.                         JmsListenerConfigUtils.JMS_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME, JmsListenerEndpointRegistry.class);
  31.             }
  32.             this.registrar.setEndpointRegistry(this.endpointRegistry);
  33.         }
  36.         // Set the custom handler method factory once resolved by the configurer
  37.         MessageHandlerMethodFactory handlerMethodFactory = this.registrar.getMessageHandlerMethodFactory();
  38.         if (handlerMethodFactory != null) {
  39.             this.messageHandlerMethodFactory.setMessageHandlerMethodFactory(handlerMethodFactory);
  40.         }
  42.         // Actually register all listeners
  43.         this.registrar.afterPropertiesSet();
  44.     }

  • 关注最后一行this.registrar.afterPropertiesSet()

    1.     @Override
    2.     public void afterPropertiesSet() {
    3.         registerAllEndpoints();
    4.     }
    6.     protected void registerAllEndpoints() {
    7.         Assert.state(this.endpointRegistry != null, "No JmsListenerEndpointRegistry set");
    8.         synchronized (this.mutex) {
    9.             for (JmsListenerEndpointDescriptor descriptor : this.endpointDescriptors) {
    10.                 // 注册监听
    11.                 this.endpointRegistry.registerListenerContainer(
    12.                         descriptor.endpoint, resolveContainerFactory(descriptor));
    13.             }
    14.             this.startImmediately = true;  // trigger immediate startup
    15.         }
    16.     }

  • 注册监听在下面分析会讲详见下文

postProcessAfterInitialization

  1.     @Override
  2.     public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
  3.         if (bean instanceof AopInfrastructureBean || bean instanceof JmsListenerContainerFactory ||
  4.                 bean instanceof JmsListenerEndpointRegistry) {
  5.             // Ignore AOP infrastructure such as scoped proxies.
  6.             return bean;
  7.         }
  9.         // 获取 bean 的代理对象.class
  10.         Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
  11.         if (!this.nonAnnotatedClasses.contains(targetClass)) {
  12.             Map<Method, Set<JmsListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
  13.                     (MethodIntrospector.MetadataLookup<Set<JmsListener>>) method -> {
  14.                         Set<JmsListener> listenerMethods = AnnotatedElementUtils.getMergedRepeatableAnnotations(
  15.                                 method, JmsListener.class, JmsListeners.class);
  16.                         return (!listenerMethods.isEmpty() ? listenerMethods : null);
  17.                     });
  18.             if (annotatedMethods.isEmpty()) {
  19.                 this.nonAnnotatedClasses.add(targetClass);
  20.                 if (logger.isTraceEnabled()) {
  21.                     logger.trace("No @JmsListener annotations found on bean type: " + targetClass);
  22.                 }
  23.             } else {
  24.                 // Non-empty set of methods
  25.                 annotatedMethods.forEach((method, listeners) ->
  26.                         listeners.forEach(listener -> processJmsListener(listener, method, bean)));
  27.                 if (logger.isDebugEnabled()) {
  28.                     logger.debug(annotatedMethods.size() + " @JmsListener methods processed on bean '" + beanName +
  29.                             "': " + annotatedMethods);
  30.                 }
  31.             }
  32.         }
  33.         return bean;
  34.     }
  1.     protected void processJmsListener(JmsListener jmsListener, Method mostSpecificMethod, Object bean) {
  2.         Method invocableMethod = AopUtils.selectInvocableMethod(mostSpecificMethod, bean.getClass());
  4.         // 设置 监听方法信息
  5.         MethodJmsListenerEndpoint endpoint = createMethodJmsListenerEndpoint();
  6.         endpoint.setBean(bean);
  7.         endpoint.setMethod(invocableMethod);
  8.         endpoint.setMostSpecificMethod(mostSpecificMethod);
  9.         endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
  10.         endpoint.setEmbeddedValueResolver(this.embeddedValueResolver);
  11.         endpoint.setBeanFactory(this.beanFactory);
  12.         endpoint.setId(getEndpointId(jmsListener));
  13.         endpoint.setDestination(resolve(jmsListener.destination()));
  14.         if (StringUtils.hasText(jmsListener.selector())) {
  15.             endpoint.setSelector(resolve(jmsListener.selector()));
  16.         }
  17.         if (StringUtils.hasText(jmsListener.subscription())) {
  18.             endpoint.setSubscription(resolve(jmsListener.subscription()));
  19.         }
  20.         if (StringUtils.hasText(jmsListener.concurrency())) {
  21.             endpoint.setConcurrency(resolve(jmsListener.concurrency()));
  22.         }
  24.         JmsListenerContainerFactory<?> factory = null;
  25.         String containerFactoryBeanName = resolve(jmsListener.containerFactory());
  26.         if (StringUtils.hasText(containerFactoryBeanName)) {
  27.             Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name");
  28.             try {
  29.                 factory = this.beanFactory.getBean(containerFactoryBeanName, JmsListenerContainerFactory.class);
  30.             } catch (NoSuchBeanDefinitionException ex) {
  31.                 throw new BeanInitializationException("Could not register JMS listener endpoint on [" +
  32.                         mostSpecificMethod + "], no " + JmsListenerContainerFactory.class.getSimpleName() +
  33.                         " with id '" + containerFactoryBeanName + "' was found in the application context", ex);
  34.             }
  35.         }
  37.         // 注册监听点 到 JmsListenerContainerFactory
  38.         this.registrar.registerEndpoint(endpoint, factory);
  39.     }
  • 将监听点注册的重要方法 org.springframework.jms.config.JmsListenerEndpointRegistrar#registerEndpoint(org.springframework.jms.config.JmsListenerEndpoint, org.springframework.jms.config.JmsListenerContainerFactory<?>)
  1.     public void registerEndpoint(JmsListenerEndpoint endpoint, @Nullable JmsListenerContainerFactory<?> factory) {
  2.         Assert.notNull(endpoint, "Endpoint must not be null");
  3.         Assert.hasText(endpoint.getId(), "Endpoint id must be set");
  5.         // Factory may be null, we defer the resolution right before actually creating the container
  6.         // jms 监听点描述
  7.         JmsListenerEndpointDescriptor descriptor = new JmsListenerEndpointDescriptor(endpoint, factory);
  9.         synchronized (this.mutex) {
  10.             if (this.startImmediately) {  // register and start immediately
  11.                 Assert.state(this.endpointRegistry != null, "No JmsListenerEndpointRegistry set");
  12.                 // 注册
  13.                 this.endpointRegistry.registerListenerContainer(descriptor.endpoint,
  14.                         resolveContainerFactory(descriptor), true);
  15.             }
  16.             else {
  17.                 this.endpointDescriptors.add(descriptor);
  18.             }
  19.         }
  20.     }

  • org.springframework.jms.config.JmsListenerEndpointRegistry#registerListenerContainer(org.springframework.jms.config.JmsListenerEndpoint, org.springframework.jms.config.JmsListenerContainerFactory<?>, boolean)

    1. public void registerListenerContainer(JmsListenerEndpoint endpoint, JmsListenerContainerFactory<?> factory,
    2.                                           boolean startImmediately) {
    4.         Assert.notNull(endpoint, "Endpoint must not be null");
    5.         Assert.notNull(factory, "Factory must not be null");
    6.         String id = endpoint.getId();
    7.         Assert.hasText(id, "Endpoint id must be set");
    9.         synchronized (this.listenerContainers) {
    10.             if (this.listenerContainers.containsKey(id)) {
    11.                 throw new IllegalStateException("Another endpoint is already registered with id '" + id + "'");
    12.             }
    13.             // 创建消息监听容器
    14.             MessageListenerContainer container = createListenerContainer(endpoint, factory);
    15.             this.listenerContainers.put(id, container);
    16.             if (startImmediately) {
    17.                 // 启动消息监听容器
    18.                 startIfNecessary(container);
    19.             }
    20.         }
    21.     }

  • org.springframework.jms.config.JmsListenerEndpointRegistry#createListenerContainer

  1.     /**
  2.      * Create and start a new container using the specified factory.
  3.      * 创建监听容器
  4.      */
  5.     protected MessageListenerContainer createListenerContainer(JmsListenerEndpoint endpoint,
  6.                                                                JmsListenerContainerFactory<?> factory) {
  8.         // 创建监听 容器
  9.         MessageListenerContainer listenerContainer = factory.createListenerContainer(endpoint);
  11.         if (listenerContainer instanceof InitializingBean) {
  12.             try {
  13.                 // 后置方法
  14.                 ((InitializingBean) listenerContainer).afterPropertiesSet();
  15.             } catch (Exception ex) {
  16.                 throw new BeanInitializationException("Failed to initialize message listener container", ex);
  17.             }
  18.         }
  20.         int containerPhase = listenerContainer.getPhase();
  21.         if (containerPhase < Integer.MAX_VALUE) {  // a custom phase value
  22.             if (this.phase < Integer.MAX_VALUE && this.phase != containerPhase) {
  23.                 throw new IllegalStateException("Encountered phase mismatch between container factory definitions: " +
  24.                         this.phase + " vs " + containerPhase);
  25.             }
  26.             this.phase = listenerContainer.getPhase();
  27.         }
  29.         return listenerContainer;
  30.     }

  • 关键接口JmsListenerContainerFactory<C extends MessageListenerContainer>

    1. public interface JmsListenerContainerFactory<C extends MessageListenerContainer> {
    3.     /**
    4.      * Create a {@link MessageListenerContainer} for the given {@link JmsListenerEndpoint}.
    5.      * 创建肩痛容器
    6.      * @param endpoint the endpoint to configure
    7.      * @return the created container
    8.      */
    9.     C createListenerContainer(JmsListenerEndpoint endpoint);
    11. }

    image-20200304092154712

  • 注册完成后是否立即启动

    1.         this.listenerContainers.put(id, container);
    2.             if (startImmediately) {
    3.                 // 启动消息监听容器
    4.                 startIfNecessary(container);
    5.             }
    7.     private void startIfNecessary(MessageListenerContainer listenerContainer) {
    8.         if (this.contextRefreshed || listenerContainer.isAutoStartup()) {
    9.             listenerContainer.start();
    10.         }
    11.     }
    • 具体实现: org.springframework.jms.listener.AbstractJmsListeningContainer#start

  • 执行完start方法就结束了processJmsListener的调用链路, postProcessAfterInitialization 也结束了

JmsListenerEndpointRegistry

  • 这个类辅助JmsListenerAnnotationBeanPostProcessor 处理

registerListenerContainer

  1.     /**
  2.      * Create a message listener container for the given {@link JmsListenerEndpoint}.
  3.      * <p>This create the necessary infrastructure to honor that endpoint
  4.      * with regards to its configuration.
  5.      * <p>The {@code startImmediately} flag determines if the container should be
  6.      * started immediately.
  7.      * <p>
  8.      * 注册监听容器
  9.      *
  10.      * @param endpoint         the endpoint to add
  11.      *                         监听点
  12.      * @param factory          the listener factory to use
  13.      *                         监听容器工厂
  14.      * @param startImmediately start the container immediately if necessary
  15.      *                         是否立即启动容器
  16.      * @see #getListenerContainers()
  17.      * @see #getListenerContainer(String)
  18.      */
  19.     public void registerListenerContainer(JmsListenerEndpoint endpoint, JmsListenerContainerFactory<?> factory,
  20.                                           boolean startImmediately) {
  22.         Assert.notNull(endpoint, "Endpoint must not be null");
  23.         Assert.notNull(factory, "Factory must not be null");
  24.         String id = endpoint.getId();
  25.         Assert.hasText(id, "Endpoint id must be set");
  27.         synchronized (this.listenerContainers) {
  28.             if (this.listenerContainers.containsKey(id)) {
  29.                 throw new IllegalStateException("Another endpoint is already registered with id '" + id + "'");
  30.             }
  31.             // 创建消息监听容器
  32.             MessageListenerContainer container = createListenerContainer(endpoint, factory);
  33.             this.listenerContainers.put(id, container);
  34.             if (startImmediately) {
  35.                 // 启动消息监听容器
  36.                 startIfNecessary(container);
  37.             }
  38.         }
  39.     }