带有 IBM MQ 的 Apache Camel [英] Apache Camel with IBM MQ

问题描述

您好，有人曾将 Camel 与 IBM 的 MQ 结合使用.我们正在考虑可能同时使用这两种产品，但没有这两种产品一起使用的示例.

Hello has anyone ever used Camel with IBM's MQ. We are looking at possibly using the two products together but have no example of the two products working together.

推荐答案

下面记录了我所能得到的最好的，说明为一个 Spring XML 应用程序上下文，它本身托管 CAMEL 上下文和路由.此示例与 IBM 本机 MQ JCA 兼容的资源适配器 v7.5、CAMEL 2.16、Spring core 4.2 一起使用.我已经将它部署在 Glassfish、Weblogic 和 JBoss EAP7 服务器中.

The best I have been able to get is documented below, illustrated as a Spring XML application context that itself hosts the CAMEL context and routes. This sample works with the IBM native MQ JCA-compliant resource adapter v7.5, CAMEL 2.16, Spring core 4.2. I have deployed it in Glassfish, Weblogic, and JBoss EAP7 servers.

复杂性必然会处理 MQ 报告的流程，其理念与普通 JMS 回复消息的理念相冲突.详细解释请参考使用 CoD over Camel JMS 组件实现本机 websphere MQ

The complexity is bound to handling the flow of MQ reports whose philosophy conflicts with that of a plain JMS reply-to message. For a detailed explanation, please refer to Implementing native websphere MQ with CoD over Camel JMS component

这个基于 CAMEL XML DSL 的示例是独立的，易于测试.

This example based on the CAMEL XML DSL is self-contained and easy to test.

我们从 Spring &CAMEL 声明:

We start with Spring & CAMEL declarations:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" 
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:util="http://www.springframework.org/schema/util" 
xmlns:context="http://www.springframework.org/schema/context" 
xmlns:camel="http://camel.apache.org/schema/spring"
xmlns:jee="http://www.springframework.org/schema/jee"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd   
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
    http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
    http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
    http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-3.0.xsd">

CAMEL 上下文遵循 2 条路线:MQ 到 JMS 和 JMS 到 MQ，这里链接起来形成一个桥梁以简化测试.

The CAMEL context follows with 2 routes: MQ to JMS and JMS to MQ, here chained to form a bridge to ease testing.

<camel:camelContext id="mqBridgeCtxt">
<camel:route id="mq2jms" autoStartup="true">

奇怪:在 Weblogic 上，获取(例如)3 个侦听器的唯一方法是强制执行 3 个连接(依次使用 3 个 Camel:from 语句)，每个连接最多 1 个会话，否则会出现 MQ 错误:MQJCA1018:只有一个会话允许每个连接.在 JBoss 上，您可以简单地调整 concurrentConsumers=...

Weird: on Weblogic, the only way to get (e.g.) 3 listeners is to enforce 3 connections (with 3 Camel:from statements in sequence) with max 1 session each, otherwise an MQ error ensues: MQJCA1018: Only one session per connection is allowed. On JBoss, you can simply adjust concurrentConsumers=...

  <camel:from uri="wmq:queue:TEST.Q1?concurrentConsumers=1&amp;disableReplyTo=true&amp;
        acknowledgementModeName=SESSION_TRANSACTED"/> 

上面的 disable disableReplyTo 选项确保 CAMEL 在我们测试 MQ 消息类型为 1=Request(-reply) 或 8=datagram(一种方式！)之前不会产生回复.此处未说明测试和回复结构.

The disable disableReplyTo option above ensures that CAMEL will not produce a reply before we can test the MQ message type to be 1=Request(-reply) or 8=datagram (one way!). That test and reply construction is not illustrated here.

然后我们在下一次发布到普通 JMS 时将 EIP 强制为 InOnly，以与 Inbound MQ 模式保持一致.

Then we enforce the EIP to InOnly on the next posting to plain JMS to be consistent with the Inbound MQ mode.

  <camel:setExchangePattern pattern="InOnly"/>
  <!-- camel:process ref="reference to your MQ message processing bean fits here" / -->
  <camel:to uri="ref:innerQueue" />
</camel:route>

这结束了 MQ 到 jms 的路由；接下来是 jms-to-MQ 路由仍然在相同的 CAMEL 上下文中:

This ends the MQ-to-jms route; next comes the jms-to-MQ route still in the same CAMEL context:

<camel:route id="jms2mq"  autoStartup="true">
  <camel:from uri="ref:innerQueue" />
  <!-- remove inner message headers and properties to test without inbound side effects! -->
  <camel:removeHeaders pattern="*"/> 
  <camel:removeProperties pattern="*" />
  <!-- camel:process ref="reference to your MQ message preparation bean fits here" / -->

现在是远程目标返回的 MQ CoD 报告的请求标志.我们还强制 MQ 消息为数据报类型(值 8).

Now comes the request flag for the MQ CoD report to be returned by remote destination. We also enforce the MQ message to be of Datagram type (value 8).

  <camel:setHeader headerName="JMS_IBM_Report_COD"><camel:simple resultType="java.lang.Integer">2048</camel:simple></camel:setHeader>
  <camel:setHeader headerName="JMS_IBM_Report_Pass_Correl_ID"><camel:simple resultType="java.lang.Integer">64</camel:simple></camel:setHeader>
  <camel:setHeader headerName="JMS_IBM_MsgType"><camel:simple resultType="java.lang.Integer">8</camel:simple></camel:setHeader>

ReplyTo 队列可以通过 ReplyTo uri 选项指定，也可以作为下面的标头指定.

The ReplyTo queue can be specified either via the ReplyTo uri option, else as a header as below.

接下来，我们使用 CamelJmsDestinationName 头来强制抑制 JMS MQ 消息头 MQRFH2(使用 targetClient MQ URL 选项值 1).换句话说，我们想发送一个普通的 MQ 二进制消息(即只有 MQMD 消息描述符后跟有效负载).

Next we do use CamelJmsDestinationName header to enforce suppressing of the JMS MQ message header MQRFH2 (using targetClient MQ URL option value 1). In other words, we want to send a plain vanilla MQ binary message (i.e. Only the MQMD message descriptor followed by the payload).

  <camel:setHeader headerName="JMSReplyTo"><camel:constant>TEST.REPLYTOQ</camel:constant></camel:setHeader>
  <camel:setHeader headerName="CamelJmsDestinationName"> <camel:constant>queue://MYQMGR/TEST.Q2?targetClient=1</camel:constant></camel:setHeader>

更多 MQMD 字段可以通过保留的 JMS 属性来控制，如下图所示.请参阅 IBM 文档中的限制.

More MQMD fields may be controlled through reserved JMS properties as illustrated below. See restrictions in IBM doc.

  <camel:setHeader headerName="JMS_IBM_Format"><camel:constant>MQSTR   </camel:constant></camel:setHeader>
  <camel:setHeader headerName="JMSCorrelationID"><camel:constant>_PLACEHOLDER_24_CHARS_ID_</camel:constant></camel:setHeader>

URI 中的目标队列被上面的 CamelJmsDestinationName 覆盖，因此 URI 中的队列名称成为占位符.

The destination queue in the URI is overwritten by the CamelJmsDestinationName above, hence the queue name in the URI becomes a placeholder.

根据观察，URI 选项 preserveMessageQos 允许发送设置了回复数据的消息(以获取 MQ CoD 报告)，但阻止 CAMEL 通过强制执行 InOnly MEP 来实例化回复消息侦听器.

The URI option preserveMessageQos is the one that - as observed - allows sending a message with the ReplyTo data being set (to get the MQ CoD Report), yet prevent CAMEL to instantiate a Reply message listener by enforcing the InOnly MEP.

  <camel:to uri="wmq:queue:PLACEHOLDER.Q.NAME?concurrentConsumers=1&amp;
            exchangePattern=InOnly&amp;preserveMessageQos=true&amp;
            includeSentJMSMessageID=true" />
</camel:route>
</camel:camelContext>

我们还没有完成，我们还需要为本地 JMS 提供程序和 Websphere MQ(通过本地 IBM WMQ JCA 资源适配器)声明队列工厂，以便根据您的上下文进行调整.我们在这里使用 JNDI 查找管理对象.

We have not finished, we have still to declare our queue factories for both a native JMS provider and Websphere MQ (via the native IBM WMQ JCA Resource Adapter), to be adjusted to your context. We use here JNDI lookups on administrative objects.

<camel:endpoint id="innerQueue" uri="jmsloc:queue:transitQueue">
</camel:endpoint>

<jee:jndi-lookup id="mqQCFBean" jndi-name="jms/MYQMGR_QCF"/>
<jee:jndi-lookup id="jmsraQCFBean" jndi-name="jms/jmsra_QCF"/>

<bean id="jmsloc" class="org.apache.camel.component.jms.JmsComponent">
  <property name="connectionFactory" ref="jmsraQCFBean" />
</bean>

<bean id="wmq" class="org.apache.camel.component.jms.JmsComponent">
  <property name="connectionFactory" ref="mqQCFBean" />
</bean>

</beans>

从 JNDI 获取工厂(和 JCA 适配器)的替代方法是将 JMS 客户端声明为 Spring bean.在 Weblogic 和 Glassfish 中，通过部署本机 IBM JCA 资源适配器并创建 JNDI 资源，然后在上面的 Spring 上下文中引用，您会得到更好的启发，在 JBoss 中，直接 MQ 客户端 bean 声明最适合如下)

An alternative to fetching the factories (and JCA adapters) from JNDI is to declare the JMS client as Spring beans. In Weblogic and Glassfish, you'll be better inspired by deploying the native IBM JCA resource adapter and creating JNDI resources then referenced in the Spring Context as above, in JBoss a direct MQ client bean declaration suits best as below)

<bean id="mqCFBean" class="com.ibm.mq.jms.MQXAConnectionFactory">
    <property name="hostName" value="${mqHost}"/>
    <property name="port" value="${mqPort}"/>
    <property name="queueManager" value="${mqQueueManager}"/>
    <property name="channel" value="${mqChannel}"/>
    <property name="transportType" value="1"/> <!-- This parameter is fixed and compulsory to work with pure MQI java libraries -->
    <property name="appName" value="${connectionName}"/>
</bean>

<bean id="wmq" class="org.apache.camel.component.jms.JmsComponent">
    <property name="connectionFactory" ref="mqCFBean"/>
    <property name="transacted" value="true"/>
    <property name="acknowledgementModeName" value="AUTO_ACKNOWLEDGE"/>
</bean>

欢迎提出意见和改进.

这篇关于带有 IBM MQ 的 Apache Camel的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！