SpringBoot2.0源码分析(二):整合ActiveMQ分析

SpringBoot具体整合ActiveMQ可参考:SpringBoot2.0应用(二):SpringBoot2.0整合ActiveMQ

ActiveMQ自动注入

当项目中存在javax.jms.Messageorg.springframework.jms.core.JmsTemplate着两个类时,SpringBoot将ActiveMQ需要使用到的对象注册为Bean,供项目注入使用。一起看一下JmsAutoConfiguration类。

@Configuration
@ConditionalOnClass({ Message.class, JmsTemplate.class })
@ConditionalOnBean(ConnectionFactory.class)
@EnableConfigurationProperties(JmsProperties.class)
@Import(JmsAnnotationDrivenConfiguration.class)
public class JmsAutoConfiguration {

    @Configuration
    protected static class JmsTemplateConfiguration {
        ......
        @Bean
        @ConditionalOnMissingBean
        @ConditionalOnSingleCandidate(ConnectionFactory.class)
        public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
            PropertyMapper map = PropertyMapper.get();
            JmsTemplate template = new JmsTemplate(connectionFactory);
            template.setPubSubDomain(this.properties.isPubSubDomain());
            map.from(this.destinationResolver::getIfUnique).whenNonNull()
                    .to(template::setDestinationResolver);
            map.from(this.messageConverter::getIfUnique).whenNonNull()
                    .to(template::setMessageConverter);
            mapTemplateProperties(this.properties.getTemplate(), template);
            return template;
        }
            ......
    }

    @Configuration
    @ConditionalOnClass(JmsMessagingTemplate.class)
    @Import(JmsTemplateConfiguration.class)
    protected static class MessagingTemplateConfiguration {

        @Bean
        @ConditionalOnMissingBean
        @ConditionalOnSingleCandidate(JmsTemplate.class)
        public JmsMessagingTemplate jmsMessagingTemplate(JmsTemplate jmsTemplate) {
            return new JmsMessagingTemplate(jmsTemplate);
        }

    }

}

RabbitAutoConfiguration主要注入了jmsMessagingTemplatejmsTemplate
RabbitAutoConfiguration同时导入了RabbitAnnotationDrivenConfiguration,注入了jmsListenerContainerFactory

消息发送

以下面的发送为例:

    jmsMessagingTemplate.convertAndSend(this.queue, msg);

这个方法会先对消息进行转换,预处理,最终通过调用JmsTemplate的doSend实现消息发送的。

    protected void doSend(Session session, Destination destination, MessageCreator messageCreator)
            throws JMSException {
        Assert.notNull(messageCreator, "MessageCreator must not be null");
        MessageProducer producer = createProducer(session, destination);
        try {
            Message message = messageCreator.createMessage(session);
            doSend(producer, message);
            if (session.getTransacted() && isSessionLocallyTransacted(session)) {
                JmsUtils.commitIfNecessary(session);
            }
        }
        finally {
            JmsUtils.closeMessageProducer(producer);
        }
    }

首先创建一个MessageProducer的实例,接着将最初的org.springframework.messaging.Message转换成javax.jms.Message,再将消息委托给producer进行发送。

消息接收

先看一个消费的事例:

@Component
public class Consumer {
    @JmsListener(destination = "sample.queue")
    public void receiveQueue(String text) {
        System.out.println(text);
    }
}

SpringBoot会去解析@JmsListener,具体实现在JmsListenerAnnotationBeanPostProcessorpostProcessAfterInitialization方法。

    public Object postProcessAfterInitialization(final Object bean, String beanName) throws BeansException {
        if (!this.nonAnnotatedClasses.contains(bean.getClass())) {
            Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
            Map<Method, Set<JmsListener>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
                    (MethodIntrospector.MetadataLookup<Set<JmsListener>>) method -> {
                        Set<JmsListener> listenerMethods = AnnotatedElementUtils.getMergedRepeatableAnnotations(
                                method, JmsListener.class, JmsListeners.class);
                        return (!listenerMethods.isEmpty() ? listenerMethods : null);
                    });
            if (annotatedMethods.isEmpty()) {
                this.nonAnnotatedClasses.add(bean.getClass());
            }
            else {
                annotatedMethods.forEach((method, listeners) ->
                        listeners.forEach(listener ->
                                processJmsListener(listener, method, bean)));
            }
        }
        return bean;
    }

SpringBoot根据注解找到了使用了@JmsListener注解的方法,当监听到ActiveMQ收到的消息时,会调用对应的方法。来看一下具体怎么进行listener和method的绑定的。

    protected void processJmsListener(JmsListener jmsListener, Method mostSpecificMethod, Object bean) {
        Method invocableMethod = AopUtils.selectInvocableMethod(mostSpecificMethod, bean.getClass());
        MethodJmsListenerEndpoint endpoint = createMethodJmsListenerEndpoint();
        endpoint.setBean(bean);
        endpoint.setMethod(invocableMethod);
        endpoint.setMostSpecificMethod(mostSpecificMethod);
        endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
        endpoint.setEmbeddedValueResolver(this.embeddedValueResolver);
        endpoint.setBeanFactory(this.beanFactory);
        endpoint.setId(getEndpointId(jmsListener));
        endpoint.setDestination(resolve(jmsListener.destination()));
        if (StringUtils.hasText(jmsListener.selector())) {
            endpoint.setSelector(resolve(jmsListener.selector()));
        }
        if (StringUtils.hasText(jmsListener.subscription())) {
            endpoint.setSubscription(resolve(jmsListener.subscription()));
        }
        if (StringUtils.hasText(jmsListener.concurrency())) {
            endpoint.setConcurrency(resolve(jmsListener.concurrency()));
        }

        JmsListenerContainerFactory<?> factory = null;
        String containerFactoryBeanName = resolve(jmsListener.containerFactory());
        if (StringUtils.hasText(containerFactoryBeanName)) {
            Assert.state(this.beanFactory != null, "BeanFactory must be set to obtain container factory by bean name");
            try {
                factory = this.beanFactory.getBean(containerFactoryBeanName, JmsListenerContainerFactory.class);
            }
            catch (NoSuchBeanDefinitionException ex) {
                throw new BeanInitializationException("Could not register JMS listener endpoint on [" +
                        mostSpecificMethod + "], no " + JmsListenerContainerFactory.class.getSimpleName() +
                        " with id '" + containerFactoryBeanName + "' was found in the application context", ex);
            }
        }

        this.registrar.registerEndpoint(endpoint, factory);
    }

先设置endpoint的相关属性,再获取jmsListenerContainerFactory,最后将endpoint注册到jmsListenerContainerFactory


本篇到此结束,如果读完觉得有收获的话,欢迎点赞、关注、加公众号【贰级天災】,查阅更多精彩历史!!!

原文地址:https://www.cnblogs.com/YJTZ/p/10617410.html

时间: 2024-07-30 10:28:00

SpringBoot2.0源码分析(二):整合ActiveMQ分析的相关文章

jmeter4.0 源码编译 二次开发

准备: 1.jmeter4.0源码 - apache-jmeter-4.0_src.zip 2.IDE Eclipse - Oxygen.3 Release (4.7.3) 3.JDK - 1.8.0_122 开始: 1.新建java project 2.导入jmeter4.0源码 导入源码后项目结构: 3.Ant 下载编译依赖的jar,下载后的jar包会在 %eclipse_workspace%\jmeter_leocnx\MyJmeter\lib目录下 首先去除掉install,选择down

AFNetworking2.0源码解析&lt;二&gt;

本篇我们继续来看看AFNetworking的下一个模块 — AFURLRequestSerialization. AFURLRequestSerialization用于帮助构建NSURLRequest,主要做了两个事情: 1.构建普通请求:格式化请求参数,生成HTTP Header. 2.构建multipart请求. 分别看看它在这两点具体做了什么,怎么做的. 1.构建普通请求 A.格式化请求参数 一般我们请求都会按key=value的方式带上各种参数,GET方法参数直接加在URL上,POST方

[Android FrameWork 6.0源码学习] View的重绘过程之WindowManager的addView方法

博客首页:http://www.cnblogs.com/kezhuang/p/ 关于Activity的contentView的构建过程,我在我的博客中已经分析过了,不了解的可以去看一下 <[Android FrameWork 6.0源码学习] Window窗口类分析> 本章博客是接着上边那篇博客分析,目的是为了引出分析ViewRootImpl这个类.现在只是分析完了Window和ActivityThread的调用过程 从ActivityThread到WindowManager再到ViewRoo

HBase1.0.0源码分析之请求处理流程分析以Put操作为例(二)

HBase1.0.0源码分析之请求处理流程分析以Put操作为例(二) 1.通过mutate(put)操作,将单个put操作添加到缓冲操作中,这些缓冲操作其实就是Put的父类的一个List的集合.如下: private List<Row> writeAsyncBuffer = new LinkedList<>(); writeAsyncBuffer.add(m); 当writeAsyncBuffer满了之后或者是人为的调用backgroundFlushCommits操作促使缓冲池中的

Solr4.8.0源码分析(19)之缓存机制(二)

Solr4.8.0源码分析(19)之缓存机制(二) 前文<Solr4.8.0源码分析(18)之缓存机制(一)>介绍了Solr缓存的生命周期,重点介绍了Solr缓存的warn过程.本节将更深入的来介绍下Solr的四种缓存类型,以及两种SolrCache接口实现类. 1.SolrCache接口实现类 前文已经提到SolrCache有两种接口实现类:solr.search.LRUCache 和 solr.search.LRUCache. 那么两者具体有啥区别呢? 1.1 solr.search.LR

Solr4.8.0源码分析(21)之SolrCloud的Recovery策略(二)

Solr4.8.0源码分析(21)之SolrCloud的Recovery策略(二) 题记:  前文<Solr4.8.0源码分析(20)之SolrCloud的Recovery策略(一)>中提到Recovery有两种策略,一是PeerSync和Replication.本节将具体介绍下PeerSync策略. PeeySync是Solr的优先选择策略,每当需要进行recovery了,Solr总是会先去判断是否需要进入PeerSync,只有当PeerSync被设置为跳过或者PeerSync时候发现没符合

Solr4.8.0源码分析(22)之 SolrCloud的Recovery策略(三)

Solr4.8.0源码分析(22)之 SolrCloud的Recovery策略(三) 本文是SolrCloud的Recovery策略系列的第三篇文章,前面两篇主要介绍了Recovery的总体流程,以及PeerSync策略.本文以及后续的文章将重点介绍Replication策略.Replication策略不但可以在SolrCloud中起到leader到replica的数据同步,也可以在用多个单独的Solr来实现主从同步.本文先介绍在SolrCloud的leader到replica的数据同步,下一篇

Solr4.8.0源码分析(4)之Eclipse Solr调试环境搭建

Solr4.8.0源码分析(4)之Eclipse Solr调试环境搭建 由于公司里的Solr调试都是用远程jpda进行的,但是家里只有一台电脑所以不能jpda进行调试,这是因为jpda的端口冲突.所以只能在Eclipse 搭建Solr的环境,折腾了一小时终于完成了. 1. JDPA远程调试 搭建换完成Solr环境后,对${TOMCAT_HOME}/bin/startup.sh 最后一行进行修改,如下所示: 1 set JPDA_ADDRESS=7070 2 exec "$PRGDIR"

Solr4.8.0源码分析(8)之Lucene的索引文件(1)

Solr4.8.0源码分析(8)之Lucene的索引文件(1) 题记:最近有幸看到觉先大神的Lucene的博客,感觉自己之前学习的以及工作的太为肤浅,所以决定先跟随觉先大神的博客学习下Lucene的原理.由于觉先大神主要介绍的是Lucene3.X系的,那我就根据源码以及结合觉先大神的来学习下4.X系的.内容可能会有些变化,且加入下我个人的理解. http://www.cnblogs.com/forfuture1978/archive/2009/12/14/1623597.html 一. 基本类型