如何在使用 Spring Integration 向 JMS 目的地发送消息时合并超时？答案

【问题标题】：How to incorporate timeout while sending message to the JMS destination using Spring Integration?如何在使用 Spring Integration 向 JMS 目的地发送消息时合并超时？
【发布时间】：2015-12-04 07:37:39
【问题描述】：

我正在使用带有 Weblogic 服务器的 Spring Web 服务。在每个请求中，我使用 JMSTemplate 将 JMS 消息发送到 JMS 目的地。以下是我的spring配置。

<int:channel id="input" />

<int:channel id="output">
    <int:dispatcher failover="true" load-balancer-ref="failoverStrategy" />
</int:channel>

<bean id="failoverStrategy" class="c.x.FailoverStrategy">
    <constructor-arg index="0">
        <list>
            <ref bean="synMessageDispatcher" />
            <ref bean="failoverHandler" />
        </list>
    </constructor-arg>
</bean>

<int:channel id="xml" />

<int-xml:marshalling-transformer
    input-channel="input" output-channel="xml" marshaller="marshaller"
    result-type="StringResult" />

<int:object-to-string-transformer
    input-channel="xml" output-channel="output" />

<bean id="marshaller" class="org.springframework.oxm.jaxb.Jaxb2Marshaller">
    <property name="contextPath" value="c.x.domain" />
</bean>

<bean id="failoverHandler"
    class="org.springframework.integration.file.FileWritingMessageHandler">
    <constructor-arg index="0">
        <bean class="java.io.File">
            <constructor-arg index="0" value="./synfailed" />
        </bean>
    </constructor-arg>
    <property name="appendNewLine" value="true" />
    <property name="autoCreateDirectory" value="true" />
    <property name="charset" value="utf-8" />
    <property name="expectReply" value="false" />
</bean>

<bean id="pollableFileSource"
    class="org.springframework.integration.file.FileReadingMessageSource"
    p:filter-ref="compositeFilter">
    <constructor-arg index="0" ref="failedMessageComparator" />
    <property name="scanEachPoll" value="false" />      
    <property name="directory">
        <bean class="java.io.File">
            <constructor-arg index="0" value="./synfailed" />
        </bean>
    </property>
</bean>

<task:scheduled-tasks>
    <task:scheduled ref="synMessageDispatcher" method="run"
        fixed-rate="10000" />
</task:scheduled-tasks>

<bean id="compositeFilter"
    class="org.springframework.integration.file.filters.CompositeFileListFilter">
    <constructor-arg>
        <list>
            <bean class="org.springframework.integration.file.filters.SimplePatternFileListFilter">
                <constructor-arg index="0" value="*.msg"/>              
            </bean>         
        </list>
    </constructor-arg>
</bean>

我使用 bean 自定义“failoverStrategy”，它在某些场景下运行良好。

如果没有来自客户端的请求并且如果 JMS 发生故障，则后续请求由“failoverHandler”成功处理（它只是将消息持久化到文件系统。

如果有来自客户端的连续请求流并且如果 JMS 出现故障，那么我将在 jmsTemplate.send(new MessageCreator() 处遇到卡住的线程。由于没有超时属性，请求永远被卡住，我必须重新启动 weblogic 服务器！请建议如何解决此问题。

以下代码 sn-p 显示了我如何发送 JMS 消息

try {

        jmsTemplate.send(new MessageCreator() {

            @Override
            public javax.jms.Message createMessage(Session arg0)
                    throws JMSException {
                // TODO Auto-generated method stub
                TextMessage message = arg0.createTextMessage(text);
                // message.setIntProperty(MESSAGE_COUNT, index);

                log.debug("Sending message: " + text);

                jmsAvailable = true;

                return message;
            }
        });
    } catch (Exception e) {
        throw new MessagingException(e.getMessage());
    }

以下是我的 JMS 相关配置。

@Bean
    public JmsTemplate jmsTemplate() throws JMSException {
        JmsTemplate jmsTemplate = new JmsTemplate();
        jmsTemplate.setConnectionFactory(cachingConnectionFactory());
        jmsTemplate.setDefaultDestinationName(syn_mpg_queue);

        return jmsTemplate;
    }

    @Bean
    public CachingConnectionFactory cachingConnectionFactory()
            throws JMSException {
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory
                .setTargetConnectionFactory(connectionFactory());
        cachingConnectionFactory.setSessionCacheSize(10);
        cachingConnectionFactory.setReconnectOnException(true);
        cachingConnectionFactory.setCacheProducers(true);
        return cachingConnectionFactory;
    }

    @Bean
    public ConnectionFactory connectionFactory() throws JMSException {
        TibjmsQueueConnectionFactory targetConnectionFactory = new TibjmsQueueConnectionFactory();
        targetConnectionFactory.setServerUrl("tcp://" + jmsServerIp + ":"
                + jmsServerPort);
        return targetConnectionFactory;
    }

【问题讨论】：

标签： spring spring-integration

【解决方案1】：

如果 JMS 出现故障，那么我将在 jmsTemplate.send(new MessageCreator() 处遇到卡住的线程

如果 JMS “宕机”，你应该得到一个异常；第一步是通过使用 jstack 或类似方法进行线程转储，找出它“卡住”的位置。

如果线程卡在 weblogic JMS 客户端并且不可中断，则框架无能为力。

【讨论】：