Apache Qpid客户端对接IBM MQ 9.1.0.6队列请求应答模式异常咨询

以下是为评估IBM MQ的请求/响应消息模式尝试构建的AMQP客户端的相关概要信息。

当前IBM队列管理器版本为9.1.0.6。

AMQP通道已按以下属性配置，可通过<<host_name>>:5672访问，通道属性如下：

Attributes for COMET.AMQP_SECCLNT
name    COMET.AMQP_SECCLNT
type    AMQP
alteration_date 2021-09-06
alteration_time 08.48.37
no_external_participants    
description COMET Secure SSL Client for AMQP
disc_interval   
hb_interval 
keep_alive_interval 
max_instances   999999999
max_instances_per_client    
max_message_length  4194304
mca_user    
sharing_conversations   
ssl_client_auth REQUIRED
ssl_cipher_spec ECDHE_RSA_AES_256_CBC_SHA384
ssl_peer_name   
cert-label  
status  INACTIVE
canStart    false
canStop false

为通过该通道访问现有队列，已创建订阅/comet/interim/claim，该订阅映射到队列端点COMET.INTERIM_CLAIMQ。

客户端配置目标 #

• 发送消息后通过JMS correlation id从同一队列端点拉取返回消息
• 将JMS消息类型设置为MQSTR而非MQHRF2
• 将回复端点设置为/comet/interim/claim，即映射到COMET.INTERIM_CLAIMQ的订阅地址

客户端代码 #

package com.ibm.mq.samples.jms.qpid;

import javax.jms.JMSContext;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.TextMessage;
import org.apache.qpid.jms.JmsConnectionFactory;

public class TestMessageSender {
       public static void main(String[] args) {

              System.setProperty("javax.net.ssl.keyStore", "cometdesktopmq.jks");
              System.setProperty("javax.net.ssl.trustStore", "cometdesktopmq.jks");
              System.setProperty("javax.net.ssl.keyStorePassword", "<<password>>");
              System.setProperty("com.ibm.mq.cfg.useIBMCipherMappings", "false");
              try {
                     String requsetUri = "amqps://host_name:5672";
                     String userName = "tgusr";
                     String password = "<<password>>";
                     JmsConnectionFactory factory = new JmsConnectionFactory(userName, password, requsetUri);
                     QueueConnection conn = factory.createQueueConnection();
                     conn.start();
                     System.out.println("conn started");
                     QueueSession sess = conn.createQueueSession(false, 1);
                     System.out.println("queue session created");
                     JMSContext context = factory.createContext(userName, password);
                     System.out.println("context created");
                     Queue q = context.createQueue("/comet/interim/claim");
                     QueueSender qs = sess.createSender(q);
                     System.out.println("queue sender created");
                     TextMessage message = context.createTextMessage("Text content");
                     message.setJMSType("MQSTR");
                     String msgId = message.getJMSMessageID();
                     message.setJMSCorrelationID(msgId);
                     Queue rq = context.createQueue("/comet/interim/claim");
                     message.setJMSReplyTo(rq);
                     qs.send(message);  // Send the message
                     System.out.println("message sent");
                     QueueReceiver reciever = sess.createReceiver(rq, "JMSCorrelationID ='" + msgId + "'");
                     Message reply = reciever.receive(10*1000); // Pickup the message using the id set in correlation id field, timeout = 10 secs
                     Inspector ins = new Inspector(reply);
                     ins.showMessageType();
                     ins.showProperties();
                     ins.showMessageHeaders();
                     ins.showMessageBody();
                     conn.close();
              } catch (Exception e) {
                     e.printStackTrace();
              }
       }
}

COMET.INTERIM_CLAIMQ队列收到的消息内容如下：

RFH [1]\[1] ¸
________________________________________
MQSTR 
MQMD
StrucId:    MD  Version:    1
MsgType:    8   Expiry: -1
Encoding:   546 CodedCharSetId: 1208
Format: MQHRF2  Priority:   0
Report: 0   Feedback:   0
MsgId:  414d5120574d51543532352020202020e4c44561003c722a
CorrelId:   414d5120574d51543532352020202020b6273861033d162d
BackoutCount:   0
AccountingToken:    0531353938370000000000000000000000000000000000000000000000000006
ApplIdentityData:       ApplOriginData: 
ReplyToQ:       ReplyToQMgr:    WMQT525
Persistence:    1   UserIdentifier: mqm
PutApplType:    26  PutApplName:    WMQT525
PutDate:    09/19/2021 22:14:20
GroupId:    000000000000000000000000000000000000000000000000
MsgSeqNumber:   1   Offset: 0
MsgFlags:   0   OriginalLength: -1
Message     
RFH [1]\[1] ¸
________________________________________
MQSTR ¸
________________________________________
ü<mq_amqp><Lis>p910-006-200703</Lis><Ver>1.0</Ver><Hdr><Dur dt='boolean'>1</Dur><Pri dt='i2'>4</Pri><Ttl dt='i8'>0</Ttl><Fac xsi:nil='true'></Fac></Hdr><Prp><Mid>ID:ce297c87-f1f4-46af-8ecb-6f6eb4d97d70:1:1:1-1</Mid><To>/comet/interim/claim</To><Sub>MQSTR</Sub><Rto>/comet/interim/claim</Rto><Aet dt='i8'>0</Aet><Crt dt='i8'>1632107660940</Crt><Gsq dt='i8'>0</Gsq></Prp><Man><x-opt-jms-reply-to>0</x-opt-jms-reply-to><x-opt-jms-dest>0</x-opt-jms-dest><x-opt-jms-msg-type>5</x-opt-jms-msg-type></Man></mq_amqp> 4<jms><Rto>topic:///comet/interim/claim</Rto></jms> Text content

待解决问题（前2项为高优先级） #

• 消息Format属性仍显示为MQHRF2，未按配置转为MQSTR
• ReplyToQ属性为空，未正确填充值
• 消息体中存在To标签值为/comet/interim/claim，该标签含义是什么？
• 消息体中存在topic:///comet/interim/claim标识，代码中设置的replyTo目标为Queue，为何会被转换为topic前缀？

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