Apache Camel 与 IBM MQ [英] 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 核心 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-to-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 uri 选项指定 ReplyTo 队列，也可以作为标题指定，如下所示.

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>

可以通过保留的 JMS 属性控制更多 MQMD 字段，如下所示.请参阅 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 是一个 - 正如观察到的 - 允许发送带有正在设置的 ReplyTo 数据的消息(以获取 MQ CoD 报告)，但防止 CAMEL 通过强制执行 InOnly MEP 来实例化 Reply 消息侦听器.

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 Context 中引用，您将获得更好的启发，如上，在 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>

欢迎提出意见和改进.

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