如何使用 CachingConnectionFactory 设置重新连接值

Question

在我的应用程序中，消息到达 inque，然后被发送到输出队列。我们通过 spring 集成来做到这一点。我的要求是，如果连接到输出队列时出现问题，那么它会尝试以 30 秒的延迟重新连接 3 次，最后如果失败则记录异常。你能帮忙看看如何实现这一目标吗？我的配置文件 -

<bean id="mqQcfParent" class="com.ibm.mq.jms.MQQueueConnectionFactory">
<property name="transportType">
<util:constant static-field="com.ibm.mq.jms.JMSC.MQJMS_TP_CLIENT_MQ_TCPIP"/>
</property>
<property name="hostName" value="${mq.out.hostname}"/>
<property name="channel" value="${mq.out.channel}"/>
<property name="port" value="${mq.out.port}"/>
</bean>
<bean id="remoteConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory" ref="mqQcfParent"/>
<property name="sessionCacheSize" value="${mq.out.cacheSize}"/>
<property name="cacheProducers" value="true"/>
<property name="cacheConsumers" value="true"/>
</bean>
<bean id="inQueue" class="com.ibm.mq.jms.MQQueue">
<property name="baseQueueName" value="${mq.in.queue}"/>
</bean>
<bean id="aircraftAssignQueue" class="com.ibm.mq.jms.MQQueue">
<property name="baseQueueName" value="${mq.out.aircraftAssignQueue}"/>
</bean>
</bean>
<bean id="failureQueue" class="com.ibm.mq.jms.MQQueue">
<property name="baseQueueName" value="${mq.out.failureQueue}"/>
</bean>
<bean id="messageListenerContainerParent" class="org.springframework.jms.listener.DefaultMessageListenerContainer" abstract="true">
<property name="destination" ref="inQueue"/>
<property name="sessionTransacted" value="true"/>
<property name="maxConcurrentConsumers" value="${mq.in.max.consumer}"/>
<property name="concurrentConsumers" value="${mq.in.min.consumer}"/>
<property name="receiveTimeout" value="5000"/>
<property name="recoveryInterval" value="60000"/>
<property name="autoStartup" value="true"/>
</bean>
<bean id="messageListenerContainerCDC" parent="messageListenerContainerParent">
<property name="connectionFactory">
<bean parent="remoteConnectionFactory">
<property name="targetConnectionFactory">
<bean parent="mqQcfParent">
<property name="hostName" value="${mq.in.cdc.hostname}"/>
<property name="channel" value="${mq.in.cdc.channel}"/>
<property name="port" value="${mq.in.cdc.port}"/>
</bean>
</property>
</bean>
</property>
</bean>
<bean id="messageListenerContainerPDC" parent="messageListenerContainerParent">
<property name="connectionFactory">
<bean parent="remoteConnectionFactory">
<property name="targetConnectionFactory">
<bean parent="mqQcfParent">
<property name="hostName" value="${mq.in.pdc.hostname}"/>
<property name="channel" value="${mq.in.pdc.channel}"/>
<property name="port" value="${mq.in.pdc.port}"/>
</bean>
</property>
</bean>
</property>
</bean>
</beans>

<context:property-placeholder location="config/application.properties"/>
<import resource="jms-listener-container-config.xml"/>
<!--   Get Input Messages  -->
<int-jms:message-driven-channel-adapter id="msgInCDC" channel="toRoute" container="messageListenerContainerCDC" error-channel="errorChannel" acknowledge="transacted"/>
<int-jms:message-driven-channel-adapter id="msgInPDC" channel="toRoute" container="messageListenerContainerPDC" error-channel="errorChannel" acknowledge="transacted"/>
<!--   Route Messages Depending on Root Element  -->
<int:channel id="toAircraftAssign"/>
<int:channel id="toDiversionChange"/>
<int:channel id="toFlightCreate"/>
<int:channel id="toFlightPlanRelease"/>
<int:channel id="toGateChange"/>
<int:channel id="toPositionReport"/>
<int:channel id="toScheduleChange"/>
<int-xml:xpath-router id="flightUpdateRouter" input-channel="toRoute" default-output-channel="errorChannel" evaluate-as-string="true">
<int-xml:xpath-expression expression="name(/*)"/>
<int-xml:mapping value="AircraftAssignment" channel="toAircraftAssign"/>
<int-xml:mapping value="DiversionChangesUpdate" channel="toDiversionChange"/>
<int-xml:mapping value="CreateFlight" channel="toFlightCreate"/>
<int-xml:mapping value="FlightPlanRelease" channel="toFlightPlanRelease"/>
<int-xml:mapping value="GateChange" channel="toGateChange"/>
<int-xml:mapping value="PositionReportUpdate" channel="toPositionReport"/>
<int-xml:mapping value="ScheduleChangesUpdate" channel="toScheduleChange"/>
</int-xml:xpath-router>
<int-jms:outbound-channel-adapter id="aircraftAssignMsgOut" channel="toAircraftAssign" connection-factory="remoteConnectionFactory" destination="aircraftAssignQueue"/>
<int-jms:outbound-channel-adapter id="diversionChangeMsgOut" channel="toDiversionChange" connection-factory="remoteConnectionFactory" destination="diversionChangeQueue"/>
<int-jms:outbound-channel-adapter id="flightCreateMsgOut" channel="toFlightCreate" connection-factory="remoteConnectionFactory" destination="flightCreateQueue"/>
<int-jms:outbound-channel-adapter id="flightPlanReleaseMsgOut" channel="toFlightPlanRelease" connection-factory="remoteConnectionFactory" destination="flightPlanReleaseQueue"/>
<int-jms:outbound-channel-adapter id="gateChangeMsgOut" channel="toGateChange" connection-factory="remoteConnectionFactory" destination="gateChangeQueue"/>
<int-jms:outbound-channel-adapter id="positionReportMsgOut" channel="toPositionReport" connection-factory="remoteConnectionFactory" destination="positionReportQueue"/>
<int-jms:outbound-channel-adapter id="scheduleChangeMsgOut" channel="toScheduleChange" connection-factory="remoteConnectionFactory" destination="scheduleChangeQueue"/>
<!--   Error Handling  -->
<int-jms:outbound-channel-adapter id="errMsgOut" channel="errorChannel" connection-factory="remoteConnectionFactory" destination="failureQueue"/>
<!--  Logger -->
<int:wire-tap pattern="to*" order="7" channel="wireTapChannel"/>
<int:logging-channel-adapter id="wireTapChannel" level="debug" logger-name="WIRETAP"/>
<!--  Logger  -->
<!--
 	<int:wire-tap pattern="to*" order="0" channel="loggerChannel"/> 
-->
<!--
 	<int:logging-channel-adapter id="loggerChannel" level="DEBUG" expression="'YYYY'"/> 
-->
</beans>

Answer 1

您没有显示任何 Spring Integration 配置，但假设您使用的是 JMS 出站通道适配器，您可以添加一个带有适当配置的 SimplyRetryPolicy 的 retry advice。

但是，如果同一代理也用于入站队列，则该会话将被中断，并且无论如何都会重新传递消息；所以你最好在代理中设置重试策略。