自定義JmsListenerContainerFactory時,containerFactory字段解讀

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

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

自定義JmsListenerContainerFactory時,containerFactory字段

最近項目中利用ActiveMQ作為消息中間件，JMS進行消息傳遞。在對@JmsListener注解研究時對“containerFactory”字段的填值產生了一些疑問。分享一下我的心得體會。

疑問

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

@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;
? ? }
}

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

按照@Bean注解的定義，在@Bean不自定義的情況下，以方法名作為標識，而若想要jmsListenerContainerQueue()被調用，containerFactory的對應字段應填寫“jmsListenerContainerQueue”而不是“jmsListenerContainerFactory”。

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

問題解析

翻找了一圈源碼后，終于發(fā)現了被“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類中的一個方法，用于生成默認的JMS監(jiān)聽工廠。那么為何調用默認的工廠生成方法，最后返回的卻是自定義的工廠？

其中最關鍵的技術實現基礎，就是springboot的scope屬性即默認的singleton模式以及@Bean注解的方法參數依賴。

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

? ? @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修飾，最終返回一個DefaultJmsListenerContainerFactoryConfigurer。

第一個關鍵點來了，項目中自定義的jmsListenerContainerQueue()中的configurer參數，同樣由于@Bean的參數依賴，由jmsListenerContainerFactoryConfigurer()生成。因此，無論containerFactory字段填“jmsListenerContainerFactory”或“jmsListenerContainerQueue”，最終獲得的DefaultJmsListenerContainerFactoryConfigurer是由同一個方法生成的。

但是問題又來了，jmsListenerContainerFactoryConfigurer()返回的DefaultJmsListenerContainerFactoryConfigurer對象，是每次新new出來的，為何兩處拿到的對象是同一個呢。

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

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

心得

這一次的疑問，使我對springboot的加載機制及JMS消息傳遞都有了深入的理解，還是那句話，實踐出真知。

springboot自定義JMSListener.destination

情景

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

開始工作

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

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

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

赫然看見救命注釋@see JmsListenerConfigurer。后面百度重啟這些小白操作被大風吹跑了。

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

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

代碼貼出來

@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("獲取本機端口號異常", 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);
    }
}

入隊的地址替換就很簡單了，注入直接換就行：

@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);
? ? ? ? ? ? }
? ? ? ? }
? ? }
}