动态加载 Spring Integration 组件答案

【问题标题】：Dynamically loading Spring Integration components动态加载 Spring Integration 组件
【发布时间】：2023-03-22 12:28:01
【问题描述】：

我需要能够将转换器添加到由 MVC 应用程序管理的正在运行的 Spring 集成应用程序上下文中，为了实现这一点，我从一个简单的 POC 开始，我在运行时加载了一个新的转换器（这个转换器不是启动时扫描），这个POC作为一个独立的java应用程序运行，只包含spring集成库，这里是主类。

@Component
public class DynamicApplication {

    @Autowired
    private ConfigurableApplicationContext appContext;

    @Autowired
    private BeanFactory beanFactory;

    private Map<String, String> channelsMap = new HashMap<String, String>();

    private Map<String, String> inboundGWMap = new HashMap<String, String>();

    public static void main(String a[]) throws Exception{
        DynamicApplication dynamicApplication = new DynamicApplication();
        GenericXmlApplicationContext context = dynamicApplication.setupContext();

        DynamicApplication localDynamicApplication = context.getBean(DynamicApplication.class);

        //then we wait for input to add new classes
        final Scanner scanner = new Scanner(System.in);

        while (true) {

            final String input = scanner.nextLine();

            if("q".equals(input.trim())) {
                break;
            }
            else{
                //in the case we got a package definition to be scanned
                localDynamicApplication.addClassesFromAnnotatedPackage(input);
                }
        }
    }

    public  GenericXmlApplicationContext setupContext() {
        final GenericXmlApplicationContext context = new GenericXmlApplicationContext();
        context.load("configuration/inbound-grand-central-configuration.xml");
        context.registerShutdownHook();
        context.refresh();
        DynamicApplication dynamicApplication = context.getBean(DynamicApplication.class);
        for(int i = 0 ; i < 1 ; i++){
            TcpInboundGateway listener = dynamicApplication.createTcpInboundGateway("server" + i, 9877 + i);
            listener.start();
        }

        return context;
    }


}

一个非常简单的服务激活器

package com.client.connector.inbound.config;

@MessageEndpoint
public class BussinesService {

    @ServiceActivator(inputChannel="toBSChannel")
    public String processIncomingMessage(String message) {
        System.out.println("Bussines logic: " + message);

        return "respondemos:" + message;

    }

}

也是一个初始转换器包 com.client.connector.inbound.config;

@MessageEndpoint
public class ISOConverter {

    @Transformer(inputChannel="iSOConverterChannel", outputChannel="toBSChannel")
    public String iSOConverter(Object payload) throws Exception {
        System.out.println("ISO Conversion place");
        if (payload instanceof byte[]) {
            return new String((byte[]) payload);
        } 
        else if (payload instanceof char[]) {
            return new String((char[]) payload);
        } 
        else { 
            return payload.toString();
        } 
    }

    @Transformer(inputChannel="iSOConverterChannel", outputChannel="toBSChannel")
    public String iSOConverter(byte[] payload) throws Exception {
        System.out.println("ISO Conversion place");
        if (payload instanceof byte[]) {
            return new String((byte[]) payload);
        } 
        else { 
            return payload.toString();
        } 
    }

}

和xml配置文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    ..............................
    ............................">
    <description>Inbound Connectors common configuration file.
    </description>
    <int:annotation-config />
    <context:component-scan base-package="com.client.connector.inbound" />
</beans>

然后我有另一个转换器在启动时没有被扫描。

package com.client.inboundtest.newclass;

@MessageEndpoint
public class DinamicAddedConverter {

    @Transformer(inputChannel="iSOConverterChannel", outputChannel="toBSChannel")
    public String iSOConverter2(Object payload) throws Exception {
        System.out.println("ISO conversion from new class");
        if (payload instanceof byte[]) {
            return new String((byte[]) payload);
        } 
        else if (payload instanceof char[]) {
            return new String((char[]) payload);
        } 
        else { 
            return payload.toString();
        } 
    }

    @Transformer(inputChannel="iSOConverterChannel", outputChannel="toBSChannel")
    public String iSOConverter2(byte[] payload) throws Exception {
        System.out.println("ISO conversion from new class");
        if (payload instanceof byte[]) {
            return new String((byte[]) payload);
        } 
        else { 
            return payload.toString();
        } 
    }

}

当我启动程序并使用 telnet 连接到端口 9877 并设置单词 test 时，我在 java 应用程序的输出中得到了这个

ISO Conversion place
Bussines logic: test
ISO Conversion place
Bussines logic: test

如果我在 java 控制台中编写 com.client.inboundtest.newclass 包，它会加载新的转换器，并显示日志

com.client.inboundtest.newclass
Added class com.client.inboundtest.newclass.DinamicAddedConverter
Nov 11, 2015 7:41:51 PM org.springframework.integration.endpoint.EventDrivenConsumer logComponentSubscriptionEvent
INFO: Adding {transformer:com.client.inboundtest.newclass.DinamicAddedConverter.iSOConverter2.transformer} as a subscriber to the 'iSOConverterChannel' channel
Nov 11, 2015 7:41:51 PM org.springframework.integration.channel.AbstractSubscribableChannel adjustCounterIfNecessary
INFO: Channel 'org.springframework.context.support.GenericXmlApplicationContext@4d9cad9d.iSOConverterChannel' has 3 subscriber(s).
Nov 11, 2015 7:41:51 PM org.springframework.integration.endpoint.AbstractEndpoint start
INFO: started com.client.inboundtest.newclass.DinamicAddedConverter.iSOConverter2.transformer
Nov 11, 2015 7:41:51 PM org.springframework.integration.endpoint.EventDrivenConsumer logComponentSubscriptionEvent
INFO: Adding {transformer} as a subscriber to the 'iSOConverterChannel' channel
Nov 11, 2015 7:41:51 PM org.springframework.integration.channel.AbstractSubscribableChannel adjustCounterIfNecessary
INFO: Channel 'org.springframework.context.support.GenericXmlApplicationContext@4d9cad9d.iSOConverterChannel' has 4 subscriber(s).
Nov 11, 2015 7:41:51 PM org.springframework.integration.endpoint.AbstractEndpoint start
INFO: started com.client.inboundtest.newclass.DinamicAddedConverter.iSOConverter2.transformer#2

向同一个打开的 telnet 连接发送新消息会在 java 应用程序控制台上提供此输出

ISO Conversion place
Bussines logic: test2
ISO Conversion place
Bussines logic: test2
ISO conversion from new class
Bussines logic: test2
ISO conversion from new class
Bussines logic: test2

这是我所期望的。

然后我尝试将我的 POC 移动到 Spring MVC 上下文中，我开始看到意外的行为，第一个意外的事情是通道不是动态创建的，因为它在第一个 POC 中发生，所以我需要定义它们像这样在上下文文件中显式显示

<beans ........">
    <context:component-scan base-package="main.package" />

    <int:annotation-config />

    <int:channel id="toBSChannel" />
    <int:channel id="iSOConverterChannel" />
    <int:channel id="restRequestChannel" />

    <bean
        class="org.springframework.web.servlet.view.InternalResourceViewResolver">
        <property name="prefix">
            <value>/WEB-INF/views/</value>
        </property>
        <property name="suffix">
            <value>.jsp</value>
        </property>
    </bean>
</beans>

我所做的是创建一个连接到我原来的 DynamicApplication 的新控制器，并将静态调用更改为由侦听器启动的 bean 上的 @PostConstruct 调用。

当应用程序启动时，它会在端口 9877 上显示我的 servlet 以及我的侦听器，当我调用刷新 servlet 时，它显示的输出与我从 java 应用程序得到的输出完全相同：

Added class com.client.inboundtest.newclass.ConverterA
Nov 11, 2015 7:55:59 PM org.springframework.integration.endpoint.EventDrivenConsumer logComponentSubscriptionEvent
INFO: Adding {transformer:com.client.inboundtest.newclass.ConverterA.iSOConverter.transformer} as a subscriber to the 'iSOConverterChannel' channel
Nov 11, 2015 7:55:59 PM org.springframework.integration.channel.AbstractSubscribableChannel adjustCounterIfNecessary
INFO: Channel 'org.springframework.web.context.WebApplicationContext:/InboundDinamicGC/dispatcher.iSOConverterChannel' has 7 subscriber(s).
Nov 11, 2015 7:55:59 PM org.springframework.integration.endpoint.AbstractEndpoint start
INFO: started com.client.inboundtest.newclass.ConverterA.iSOConverter.transformer
Nov 11, 2015 7:55:59 PM org.springframework.integration.endpoint.EventDrivenConsumer logComponentSubscriptionEvent
INFO: Adding {transformer} as a subscriber to the 'iSOConverterChannel' channel
Nov 11, 2015 7:55:59 PM org.springframework.integration.channel.AbstractSubscribableChannel adjustCounterIfNecessary
INFO: Channel 'org.springframework.web.context.WebApplicationContext:/InboundDinamicGC/dispatcher.iSOConverterChannel' has 8 subscriber(s).
Nov 11, 2015 7:55:59 PM org.springframework.integration.endpoint.AbstractEndpoint start
INFO: started com.client.inboundtest.newclass.ConverterA.iSOConverter.transformer#2

问题是消息永远不会到达新转换器，因为它发生在普通 java POC 上，有什么想法吗？

【问题讨论】：

标签： java spring-mvc dynamic spring-integration

【解决方案1】：

请查看类似的问题How to hook up a list of message driven adapters without actually writing each one out? 及其讨论，以及那里的链接。

您应该做的只是创建一个child 应用程序上下文。就像你在 setupContext() 中所做的那样，但当然是把 MVC parent 和 refresh() 推到那里。

【讨论】：

太棒了！你介意分享最终的解决方案吗？我们可以将其用作示例甚至一些文档改进。如您所见，此功能变得流行。我什至会说我们应该考虑一些开箱即用的组件......好吧，根据您的经验提出一个关于这个问题的 Jira，我们会做进一步的事情。
嗨，我刚刚上传到github.com/camicase82/camicasePOC ;)，我还有一个问题，为什么在简单的java版本中可以使用的自动创建频道功能在网页版中不起作用？我的意思是在 java 版本中不需要在上下文中，但在 web 版本中，如果我没有指定，我会得到一个找不到 bean。
我认为确实如此。但在 MVC 情况下，您有 parent/child 关系。因此，它们是在子上下文中创建的，对于您要从中发送消息的父级来说是不可见的。将它们放在parent 中可以让孩子们看到它们。
嗨，你能指点我的 Jira 网站吗？我还将进一步研究自动创建通道的事情，我只是仔细检查了应用程序，所有 SI 组件都在同一上下文中启动。