双IBM MQ主备队列自动适配监听需求及配置实现咨询

Let's break down how to modify your existing Spring configuration to achieve automatic adaptation for active-standby IBM MQ queues, so your system can handle sender switches without manual intervention.

Key Issues with Current Setup #

Your current config creates separate listeners and containers for each queue, which works but lacks automatic failover/adaptation. Also, having two distinct listeners adds unnecessary redundancy if the message processing logic is the same for both queues.

Step-by-Step Implementation #

1. Unify Message Listener

First, replace your two separate listeners with a single shared listener—since both active and standby queues should use the same business logic to process messages:

@Bean
public MQUnifiedMessageListener unifiedMessageListener() {
    return new MQUnifiedMessageListener();
}

// Unified listener to handle messages from both queues
public class MQUnifiedMessageListener implements MessageListener {
    private static final Logger log = LoggerFactory.getLogger(MQUnifiedMessageListener.class);

    @Override
    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                String content = ((TextMessage) message).getText();
                log.info("Received message from queue: {}", content);
                // Add your core business processing logic here
            }
        } catch (JMSException e) {
            log.error("Failed to process message", e);
        }
    }
}

2. Switch to DefaultMessageListenerContainer

Replace SimpleMessageListenerContainer with DefaultMessageListenerContainer—it has built-in connection recovery, auto-retry on failures, and is more robust for long-running queue listening:

@Bean
public DefaultMessageListenerContainer activeQueueContainer(MQQueueConnectionFactory mqQueueConnectionFactoryA,
                                                           MQUnifiedMessageListener listener) {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(mqQueueConnectionFactoryA);
    container.setDestinationName(queueName1);
    container.setMessageListener(listener);
    container.setAutoStartup(true);
    container.setRecoveryInterval(5000); // Retry connection every 5 seconds on failure
    container.setErrorHandler(t -> log.error("Active queue container encountered error", t));
    return container;
}

@Bean
public DefaultMessageListenerContainer standbyQueueContainer(MQQueueConnectionFactory mqQueueConnectionFactoryB,
                                                             MQUnifiedMessageListener listener) {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(mqQueueConnectionFactoryB);
    container.setDestinationName(queueName2);
    container.setMessageListener(listener);
    container.setAutoStartup(true);
    container.setRecoveryInterval(5000);
    container.setErrorHandler(t -> log.error("Standby queue container encountered error", t));
    return container;
}

3. Add Automatic Connection Factory Routing (Optional, for Failover)

If you want to prioritize the active queue manager and automatically switch to standby when it's unavailable, use AbstractRoutingConnectionFactory to dynamically route connections:

@Bean
public AbstractRoutingConnectionFactory routingConnectionFactory(MQQueueConnectionFactory mqQueueConnectionFactoryA,
                                                               MQQueueConnectionFactory mqQueueConnectionFactoryB) {
    AbstractRoutingConnectionFactory routingFactory = new AbstractRoutingConnectionFactory() {
        @Override
        protected Object determineCurrentLookupKey() {
            // Check if active queue manager is available; use it if yes, else switch to standby
            return isQueueManagerAvailable(queueManager1, host1, port1) ? "active" : "standby";
        }
    };

    Map<Object, ConnectionFactory> targetFactories = new HashMap<>();
    targetFactories.put("active", mqQueueConnectionFactoryA);
    targetFactories.put("standby", mqQueueConnectionFactoryB);

    routingFactory.setTargetConnectionFactories(targetFactories);
    routingFactory.setDefaultTargetConnectionFactory(mqQueueConnectionFactoryA); // Fallback to active
    return routingFactory;
}

// Helper method to check queue manager availability
private boolean isQueueManagerAvailable(String qmName, String host, int port) {
    MQQueueConnectionFactory testFactory = new MQQueueConnectionFactory();
    try {
        testFactory.setHostName(host);
        testFactory.setQueueManager(qmName);
        testFactory.setPort(port);
        testFactory.setChannel(channel);
        testFactory.setTransportType(WMQConstants.WMQ_CM_CLIENT);
        testFactory.setSSLCipherSuite(sslCipherSuite);

        // Attempt a quick connection test
        Connection testConn = testFactory.createConnection();
        testConn.close();
        return true;
    } catch (JMSException e) {
        log.warn("Queue Manager {} is unavailable: {}", qmName, e.getMessage());
        return false;
    }
}

4. Optimize Configuration Management (Optional)

Replace scattered @Value annotations with a unified configuration properties class for cleaner code:

@ConfigurationProperties(prefix = "spring.ibmmq")
@Data
public class MQConfigProperties {
    private String host1;
    private String host2;
    private Integer port1;
    private Integer port2;
    private String queueManager1;
    private String queueManager2;
    private String channel;
    private String queueName1;
    private String queueName2;
    private String sslCipherSuite;
}

Then inject this class into your MessageConfig instead of using @Value:

@Autowired
private MQConfigProperties mqProps;

// Update your connection factory beans to use mqProps instead of individual @Value fields
@Bean @Primary 
public MQQueueConnectionFactory mqQueueConnectionFactoryA() {
    MQQueueConnectionFactory factory = new MQQueueConnectionFactory();
    try {
        factory.setHostName(mqProps.getHost1());
        factory.setQueueManager(mqProps.getQueueManager1());
        factory.setPort(mqProps.getPort1());
        // ... other configs from mqProps
    } catch (JMSException e) {
        log.error("Failed to create active connection factory", e);
    }
    return factory;
}

How This Works #

• Dual Container Listening: Both active and standby queues are monitored simultaneously. When the sender switches to the standby queue, your system will pick up messages immediately without any manual changes.
• Automatic Recovery: DefaultMessageListenerContainer automatically retries connections if a queue manager goes down, and resumes listening once it's back online.
• Dynamic Routing (Optional): The routing connection factory ensures your system always uses the available active queue manager, falling back to standby only when needed.

内容的提问来源于stack exchange，提问作者Swati Naroliya