Spring Integration：多Message Driven Channel Adapter传值及WMQ消费适配咨询

Got it, let's break down your requirements step by step—here's how to implement multiple WMQ Message Driven Channel Adapters in Spring Integration, pass target values to a common channel, and capture adapter details for business logic:

Each adapter needs its own WMQ connection factory (to connect to different queue managers/queues), and all adapters should route messages to the same common outbound channel. Below are examples for both Java and XML configurations:

Java Configuration Example #

@Configuration
@EnableIntegration
public class WmqIntegrationConfig {

    // WMQ Connection Factory for first queue
    @Bean
    public MQQueueConnectionFactory wmqConnectionFactory1() {
        MQQueueConnectionFactory factory = new MQQueueConnectionFactory();
        factory.setHostName("wmq-host-1");
        factory.setPort(1414);
        factory.setQueueManager("QMGR1");
        factory.setChannel("CHANNEL1");
        factory.setTransportType(WMQConstants.WMQ_CM_CLIENT);
        return factory;
    }

    // WMQ Connection Factory for second queue
    @Bean
    public MQQueueConnectionFactory wmqConnectionFactory2() {
        MQQueueConnectionFactory factory = new MQQueueConnectionFactory();
        factory.setHostName("wmq-host-2");
        factory.setPort(1414);
        factory.setQueueManager("QMGR2");
        factory.setChannel("CHANNEL2");
        factory.setTransportType(WMQConstants.WMQ_CM_CLIENT);
        return factory;
    }

    // Common outbound channel shared by all adapters
    @Bean
    public MessageChannel commonOutboundChannel() {
        return new DirectChannel();
    }

    // First Message Driven Channel Adapter
    @Bean
    public MessageProducerSupport wmqMessageDrivenAdapter1() {
        JmsMessageDrivenChannelAdapter adapter = new JmsMessageDrivenChannelAdapter();
        adapter.setConnectionFactory(wmqConnectionFactory1());
        adapter.setDestinationName("QUEUE1");
        adapter.setOutputChannel(commonOutboundChannel());
        // Inject adapter identity and target value via header enricher
        adapter.setTransformer(headerEnricherForAdapter("adapter-queue1", "target-system-a"));
        return adapter;
    }

    // Second Message Driven Channel Adapter
    @Bean
    public MessageProducerSupport wmqMessageDrivenAdapter2() {
        JmsMessageDrivenChannelAdapter adapter = new JmsMessageDrivenChannelAdapter();
        adapter.setConnectionFactory(wmqConnectionFactory2());
        adapter.setDestinationName("QUEUE2");
        adapter.setOutputChannel(commonOutboundChannel());
        adapter.setTransformer(headerEnricherForAdapter("adapter-queue2", "target-system-b"));
        return adapter;
    }

    // Helper to create header enrichers for consistent context injection
    private HeaderEnricher headerEnricherForAdapter(String adapterId, String targetValue) {
        Map<String, HeaderValueMessageProcessor<?>> headers = new HashMap<>();
        headers.put("adapterId", new StaticHeaderValueMessageProcessor<>(adapterId));
        headers.put("targetValue", new StaticHeaderValueMessageProcessor<>(targetValue));
        return new HeaderEnricher(headers);
    }
}

XML Configuration Example #

<!-- Common outbound channel -->
<int:channel id="commonOutboundChannel"/>

<!-- First WMQ Adapter with header enrichment -->
<int-jms:message-driven-channel-adapter id="wmqAdapter1"
    connection-factory="wmqConnectionFactory1"
    destination-name="QUEUE1"
    output-channel="commonOutboundChannel">
    <int:header-enricher>
        <int:header name="adapterId" value="adapter-queue1"/>
        <int:header name="targetValue" value="target-system-a"/>
    </int:header-enricher>
</int-jms:message-driven-channel-adapter>

<!-- Second WMQ Adapter with header enrichment -->
<int-jms:message-driven-channel-adapter id="wmqAdapter2"
    connection-factory="wmqConnectionFactory2"
    destination-name="QUEUE2"
    output-channel="commonOutboundChannel">
    <int:header-enricher>
        <int:header name="adapterId" value="adapter-queue2"/>
        <int:header name="targetValue" value="target-system-b"/>
    </int:header-enricher>
</int-jms:message-driven-channel-adapter>

<!-- WMQ Connection Factories -->
<bean id="wmqConnectionFactory1" class="com.ibm.mq.jms.MQQueueConnectionFactory">
    <property name="hostName" value="wmq-host-1"/>
    <property name="port" value="1414"/>
    <property name="queueManager" value="QMGR1"/>
    <property name="channel" value="CHANNEL1"/>
    <property name="transportType" value="1"/> <!-- WMQ_CM_CLIENT -->
</bean>

<bean id="wmqConnectionFactory2" class="com.ibm.mq.jms.MQQueueConnectionFactory">
    <property name="hostName" value="wmq-host-2"/>
    <property name="port" value="1414"/>
    <property name="queueManager" value="QMGR2"/>
    <property name="channel" value="CHANNEL2"/>
    <property name="transportType" value="1"/> <!-- WMQ_CM_CLIENT -->
</bean>

As shown above, we use a header enricher to inject custom context (like adapterId and targetValue) into every message before it reaches the common channel. This ensures each message carries the metadata you need to route or process it correctly.

If your target value is dynamic (not static), you can use SpEL expressions in the header enricher:

<!-- Dynamic target value example in XML -->
<int:header name="dynamicTargetValue" expression="#{someBean.resolveTargetValue()}"/>

Or in Java config:

headers.put("dynamicTargetValue", new ExpressionEvaluatingHeaderValueMessageProcessor<>(
    "#{someBean.resolveTargetValue()}", String.class));

When processing messages from the common channel, extract the custom headers (or use built-in Spring Integration headers) to access adapter context. Here's how to do it in a service activator:

Example Service Activator #

@Service
public class CommonMessageProcessor {

    @ServiceActivator(inputChannel = "commonOutboundChannel")
    public void processMessage(Message<String> message) {
        // Extract custom headers we added earlier
        String adapterId = message.getHeaders().get("adapterId", String.class);
        String targetValue = message.getHeaders().get("targetValue", String.class);

        // Optional: Access built-in sender channel header (for channel context)
        MessageChannel senderChannel = message.getHeaders()
            .get(IntegrationMessageHeaderAccessor.SENDER_CHANNEL, MessageChannel.class);
        String senderChannelId = senderChannel != null ? senderChannel.getBeanName() : null;

        // Execute business logic based on adapter context
        switch(adapterId) {
            case "adapter-queue1":
                handleQueue1Messages(message.getPayload(), targetValue);
                break;
            case "adapter-queue2":
                handleQueue2Messages(message.getPayload(), targetValue);
                break;
            default:
                handleUnknownSource(message.getPayload());
        }
    }

    private void handleQueue1Messages(String payload, String targetValue) {
        // Business logic for messages from first WMQ queue
    }

    private void handleQueue2Messages(String payload, String targetValue) {
        // Business logic for messages from second WMQ queue
    }

    private void handleUnknownSource(String payload) {
        // Fallback logic for unrecognized adapters
    }
}

Key Tips #

• Custom Headers: Using explicit headers like adapterId is the most reliable way to track message origin—built-in headers might not always map directly to your adapter bean.
• Error Handling: If you need per-adapter error handling, set an error-channel on each message-driven adapter to route exceptions to a dedicated handler, even if they share the main outbound channel.

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