快捷導(dǎo)航

自定義JmsListenerContainerFactory時(shí),containerFactory字段解讀

更新時(shí)間：2023年07月07日 09:29:50 作者：土狗頭子

這篇文章主要介紹了自定義JmsListenerContainerFactory時(shí),containerFactory字段解讀，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

自定義JmsListenerContainerFactory時(shí),containerFactory字段

最近項(xiàng)目中利用ActiveMQ作為消息中間件，JMS進(jìn)行消息傳遞。在對(duì)@JmsListener注解研究時(shí)對(duì)“containerFactory”字段的填值產(chǎn)生了一些疑問(wèn)。分享一下我的心得體會(huì)。

疑問(wèn)

@Component
public class TestMessageListener implements MessageListener {
? ? @Override
? ? @JmsListener(destination = "myQueue", containerFactory = "jmsListenerContainerFactory")
? ? public void onMessage(Message message) {
? ? ? ? ...
? ? ? ? 業(yè)務(wù)代碼
? ? ? ? ...
? ? }
}

@Configuration
@EnableJms?
public class ActiveMQConfig {
? ? /**
? ? ?* 發(fā)布-訂閱模式的ListenerContainer
? ? ?*/
? ? @Bean
? ? public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ConnectionFactory factory,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? DefaultJmsListenerContainerFactoryConfigurer configurer) {
? ? ? ? DefaultJmsListenerContainerFactory jmsListenerContainerFactory = new DefaultJmsListenerContainerFactory();
? ? ? ? configurer.configure(jmsListenerContainerFactory, factory);
? ? ? ? jmsListenerContainerFactory.setConnectionFactory(factory);
? ? ? ? ...
? ? ? ? return jmsListenerContainerFactory;
? ? }
? ? /**
? ? ?* P2P模式的ListenerContainer
? ? ?*/
? ? @Bean
? ? public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ConnectionFactory factory,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? DefaultJmsListenerContainerFactoryConfigurer configurer) {
? ? ? ? DefaultJmsListenerContainerFactory jmsListenerContainerFactory = new DefaultJmsListenerContainerFactory();
? ? ? ? configurer.configure(jmsListenerContainerFactory, factory);
? ? ? ? jmsListenerContainerFactory.setConnectionFactory(factory);
? ? ? ? ...
? ? ? ? return jmsListenerContainerFactory;
? ? }
}

我查找項(xiàng)目中，containerFactory的對(duì)應(yīng)字段“jmsListenerContainerFactory”，并不存在被該字段修飾的@Bean注解，而最終項(xiàng)目運(yùn)行起來(lái)后，卻正確使用了jmsListenerContainerQueue()定義的配置屬性。

按照@Bean注解的定義，在@Bean不自定義的情況下，以方法名作為標(biāo)識(shí)，而若想要jmsListenerContainerQueue()被調(diào)用，containerFactory的對(duì)應(yīng)字段應(yīng)填寫(xiě)“jmsListenerContainerQueue”而不是“jmsListenerContainerFactory”。

對(duì)此我深感疑惑，由于對(duì)springboot項(xiàng)目也是初上手，對(duì)很多加載機(jī)制非常陌生，在翻閱了一些資料與查看了相關(guān)源碼后，終于理解了其中的奧妙。

問(wèn)題解析

翻找了一圈源碼后，終于發(fā)現(xiàn)了被“jmsListenerContainerFactory”修飾的@Bean注解

? ? @Bean
? ? @ConditionalOnMissingBean(
? ? ? ? name = {"jmsListenerContainerFactory"}
? ? )
? ? public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(DefaultJmsListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) {
? ? ? ? DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
? ? ? ? configurer.configure(factory, connectionFactory);
? ? ? ? return factory;
? ? }

這是在JMS源碼的JmsAnnotationDrivenConfiguration類中的一個(gè)方法，用于生成默認(rèn)的JMS監(jiān)聽(tīng)工廠。那么為何調(diào)用默認(rèn)的工廠生成方法，最后返回的卻是自定義的工廠？

其中最關(guān)鍵的技術(shù)實(shí)現(xiàn)基礎(chǔ)，就是springboot的scope屬性即默認(rèn)的singleton模式以及@Bean注解的方法參數(shù)依賴。

首先，jmsListenerContainerFactory()被@Bean注解，該方法又需要參數(shù)configurer，而該configurer由于@Bean注解的方法參數(shù)依賴，依賴調(diào)用了以下方法生成

? ? @Bean
? ? @ConditionalOnMissingBean
? ? public DefaultJmsListenerContainerFactoryConfigurer jmsListenerContainerFactoryConfigurer() {
? ? ? ? DefaultJmsListenerContainerFactoryConfigurer configurer = new DefaultJmsListenerContainerFactoryConfigurer();
? ? ? ? configurer.setDestinationResolver((DestinationResolver)this.destinationResolver.getIfUnique());
? ? ? ? configurer.setTransactionManager((JtaTransactionManager)this.transactionManager.getIfUnique());
? ? ? ? configurer.setMessageConverter((MessageConverter)this.messageConverter.getIfUnique());
? ? ? ? configurer.setJmsProperties(this.properties);
? ? ? ? return configurer;
? ? }

該方法同樣被@Bean修飾，最終返回一個(gè)DefaultJmsListenerContainerFactoryConfigurer。

第一個(gè)關(guān)鍵點(diǎn)來(lái)了，項(xiàng)目中自定義的jmsListenerContainerQueue()中的configurer參數(shù)，同樣由于@Bean的參數(shù)依賴，由jmsListenerContainerFactoryConfigurer()生成。因此，無(wú)論containerFactory字段填“jmsListenerContainerFactory”或“jmsListenerContainerQueue”，最終獲得的DefaultJmsListenerContainerFactoryConfigurer是由同一個(gè)方法生成的。

但是問(wèn)題又來(lái)了，jmsListenerContainerFactoryConfigurer()返回的DefaultJmsListenerContainerFactoryConfigurer對(duì)象，是每次新new出來(lái)的，為何兩處拿到的對(duì)象是同一個(gè)呢。

那么引出第二個(gè)關(guān)鍵點(diǎn)，springboot中Bean加載的singleton機(jī)制，由于該機(jī)制，導(dǎo)致被@Bean所注解的方法，在項(xiàng)目啟動(dòng)時(shí)執(zhí)行，并將該返回值放入BeanFactory中保存。后續(xù)若該方法被調(diào)用，不像普通java代碼一樣，重新執(zhí)行一遍方法獲取返回值，而是直接從BeanFactory中獲取。

由此，無(wú)論containerFactory使用哪個(gè)字段，最終獲得的Factory配置，均是同一個(gè)。

心得

這一次的疑問(wèn)，使我對(duì)springboot的加載機(jī)制及JMS消息傳遞都有了深入的理解，還是那句話，實(shí)踐出真知。

springboot自定義JMSListener.destination

情景

項(xiàng)目在組內(nèi)開(kāi)發(fā)人員電腦上經(jīng)常跑本地，activemq的隊(duì)列名寫(xiě)在配置文件上。由于代碼分支不一樣，導(dǎo)致消息經(jīng)常被不正常得消費(fèi)掉。想要改進(jìn)這個(gè)問(wèn)題，最簡(jiǎn)單的是將注解@JMSListener 改為動(dòng)態(tài)加載監(jiān)聽(tīng)BEAN，但是大家不想為了這個(gè)事改變開(kāi)發(fā)習(xí)慣，所以定為動(dòng)態(tài)修改入隊(duì)和監(jiān)聽(tīng)的地址。

開(kāi)始工作

從JMSListener注解入手，注釋中提到了JmsListenerContainerFactory和DestinationResolver。從JmsListenerContainerFactory開(kāi)始查找，項(xiàng)目中用到的默認(rèn)factory-DefaultJmsListenerContainerFactory,生產(chǎn)泛型為DefaultMessageListenerContainer的container，找到了最終監(jiān)聽(tīng)內(nèi)部類scheduledInvokers，但是沒(méi)方法修改該類的內(nèi)容。

換個(gè)路子看DestinationResolver。找到了DynamicDestinationResolver，看到了希望，查找包內(nèi)引用查到了JmsDestinationAccessor，方法是new DynamicDestinationResolver()，開(kāi)始想研究這個(gè)resolver并想辦法替換。找了一路十八開(kāi)，原來(lái)這玩意是個(gè)池子，把初始化好的放里邊，你改變這里的東西其實(shí)沒(méi)啥大用處，隊(duì)列已經(jīng)建好了，嘗試stop舊的container然后new一個(gè)新的，證明也是徒勞。期間也嘗試了EmbeddedValueResolver，發(fā)現(xiàn)更坑，直接就是spring底層的BeanFactory。

把org.springframework.jms.listener下面的包翻了一遍沒(méi)結(jié)果，就要放棄的時(shí)候，發(fā)現(xiàn)annotation包里有個(gè)JmsListenerAnnotationBeanPostProcessor一下豁然開(kāi)朗。我不能初始化之后改，為啥不在初始化之前改呢……繼續(xù)折騰。

赫然看見(jiàn)救命注釋@see JmsListenerConfigurer。后面百度重啟這些小白操作被大風(fēng)吹跑了。

總之最后結(jié)果就是，繼承JmsListenerEndpointRegistry，重寫(xiě)registerListenerContainer方法，將endpoint的destincation加上自己的suffix，然后繼續(xù)走父類邏輯，然后通過(guò)JmsListenerConfigurer替換JmsListenerEndpointRegistrar中的Registry。

最后，這東西就是為了本地啟動(dòng)有的，就不要影響線上了，排除prd環(huán)境。

代碼貼出來(lái)

@Slf4j
@Component
@Profile("!prd")
public class MyJmsListenerConfigurer implements JmsListenerConfigurer, ApplicationContextAware {
    private ConfigurableApplicationContext context;
    @Autowired
    private MyJmsListenerEndpointRegistry myJmsListenerEndpointRegistry;
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        if (applicationContext instanceof ConfigurableApplicationContext) {
            this.context = (ConfigurableApplicationContext) applicationContext;
        }
    }
    @Override
    public void configureJmsListeners(JmsListenerEndpointRegistrar registrar) {
        try {
            InetAddress address = InetAddress.getLocalHost();
            String localName = "-" + address.getHostName() + "@" + address.getHostAddress();
            myJmsListenerEndpointRegistry.setSuffix(localName);
            registrar.setEndpointRegistry(myJmsListenerEndpointRegistry);
        } catch (UnknownHostException e) {
            log.error("獲取本機(jī)端口號(hào)異常", e);
            if (context != null) {
                context.close();
            }
        }
    }
}

@Component
@Profile("!prd")
public class MyJmsListenerEndpointRegistry extends JmsListenerEndpointRegistry {
    private String suffix;
    public void setSuffix(String suffix) {
        this.suffix = suffix;
    }
    @Override
    public void registerListenerContainer(JmsListenerEndpoint endpoint, JmsListenerContainerFactory<?> factory, boolean startImmediately) {
        MethodJmsListenerEndpoint methodEndpoint = (MethodJmsListenerEndpoint) endpoint;
        methodEndpoint.setDestination(methodEndpoint.getDestination() + suffix);
        super.registerListenerContainer(endpoint, factory, startImmediately);
    }
}

入隊(duì)的地址替換就很簡(jiǎn)單了，注入直接換就行：

@Component
@Order
@Profile(value = "!prd")
public class LocalQueueBeanNameChanger implements CommandLineRunner {
? ? @Autowired
? ? private List<Queue> queueList;
? ? @Override
? ? public void run(String... args) throws Exception {
? ? ? ? InetAddress address = InetAddress.getLocalHost();
? ? ? ? String localName = "-" + address.getHostName() + "@" + address.getHostAddress();
? ? ? ? for (Queue queue : queueList) {
? ? ? ? ? ? if (queue instanceof ActiveMQQueue) {
? ? ? ? ? ? ? ? ActiveMQQueue activeMQQueue = (ActiveMQQueue) queue;
? ? ? ? ? ? ? ? activeMQQueue.setPhysicalName(activeMQQueue.getPhysicalName() + localName);
? ? ? ? ? ? }
? ? ? ? }
? ? }
}