互操作性 Azure 服务总线消息队列消息

Question

我有一个 Java 应用程序和一个 NodeJS 应用程序都使用一个 Azure 服务总线消息队列。

我在客户身上看到了一些奇怪的效果，如下所示。

JAVA MESSAGE Producer（根据 Azure JMS 教程使用 QPID 库）：

 TextMessage message = sendSession.createTextMessage();
        message.setText("Test AMQP message from JMS");
        long randomMessageID = randomGenerator.nextLong() >>>1;
        message.setJMSMessageID("ID:" + randomMessageID);
        sender.send(message);
        System.out.println("Sent message with JMSMessageID = " + message.getJMSMessageID());

输出： 发送消息 JMSMessageID = ID:2414932965987073843

NODEJS 消息消费者：

serviceBus.receiveQueueMessage(queue, {timeoutIntervalInS: timeOut, isReceiveAndDelete: true}, function(err, message) {
if(message !==null)console.log(util.inspect(message, {showHidden: false, depth: null}));
});

输出：

{ body: '@\u0006string\b3http://schemas.microsoft.com/2003/10/Serialization/�\u001aTest AMQP message from JMS',
brokerProperties:
{ DeliveryCount: 1,
EnqueuedSequenceNumber: 5000004,
EnqueuedTimeUtc: 'Wed, 04 Nov 2015 21:28:21 GMT',
MessageId: '2414932965987073843',
PartitionKey: '89',
SequenceNumber: 59672695067659070,
State: 'Active',
TimeToLive: 1209600,
To: 'moequeue' },
contentType: 'application/xml; charset=utf-8' }

如果我将其与通过 serviceBus.sendQueueMessage() 插入队列的消息进行比较，则属性如下所示：

{ body: 'test message',
brokerProperties:
{ DeliveryCount: 1,
EnqueuedSequenceNumber: 0,
EnqueuedTimeUtc: 'Wed, 04 Nov 2015 21:44:03 GMT',
MessageId: 'bc0a3d4f-15ba-434f-9fb0-1a3789885f8c',
PartitionKey: '734',
SequenceNumber: 37436171906517256,
State: 'Active',
TimeToLive: 1209600 },
contentType: 'text/plain',
customProperties:
{ message_number: 0,
sent_date: Wed Nov 04 2015 21:44:03 GMT+0000 (UTC) } }

所以内容类型一开始就不同 - 为什么？ - 然后第一个消息有效负载正文中的奇怪垃圾来自哪里：@\u0006string\b3http://schemas.microsoft.com/2003/10/Serialization/�\u001a 这是序列化的结果吗？如何缓解这种情况？

在这里也可以找到代码： http://pastebin.com/T9RTFRBk

Answer 1

我遇到了同样的问题，即以“@\u0006string\b3http://schemas.microsoft.com/2003/10/Serialization/\u0001”为前缀的消息正文。

我使用 Nodejs 以及 azure-iot-device-mqtt 和 azure-iot-device 包将消息发送到 IoT 中心。我正在使用流分析作业从 IoT 中心接收消息并将它们发布到队列。我正在使用带有 amqp10 包的 Nodejs 从队列中接收事件。

问题不是由我发送或接收消息的方式引起的。相反，问题在于流分析兼容性级别！兼容级别 1.0（至少在我部署时是默认的）使用 DataContractSerializer 将消息序列化为 XML 流！ Microsoft 使用兼容性级别 1.1 更改（修复）了此问题。因此，您可能只需将流分析作业的兼容性级别（CONFIGURE->Compatibility level）更改为 1.1。

见：https://docs.microsoft.com/en-us/azure/stream-analytics/stream-analytics-compatibility-level#major-changes-in-the-latest-compatibility-level-11：

Answer 2

我们遇到了完全相同的问题，尽管在一个使用基于 Camel 的生产者的示例中涉及更多。由于我们环境的变化，我们开始遇到这些问题。

这里的问题是 REST 服务在对节点客户端的 HTTP 响应进行编码时如何解释 JMS 消息。

我们发现 JmsTextmessage 出于某种原因（不完全清楚）被假定为“application/xml”类型，并且内容将按原样转发。因此，您在示例中得到了 OUTPUT。

如果改为使用 JmsByteMessage，则内容将被解释为“应用程序/八位字节流”并且不会在传输中被破坏。

所以尝试以下方式：

BytesMessage message = sendSession.createBytesMessage();
String body = "Test AMQP message from JMS";
message.writeBytes(body.getBytes(StandardCharsets.UTF_8));
sender.send(message);

我们使用它来传输 JSON 编码数据以由 Node.js 客户端解释。

Answer 3

Azure 服务总线支持两种不同的协议：AMQP 和 HTTP。使用 qpid 库的 Java/JMS 将 AMQP 协议用于 ServiceBus。但是，ServiceBus REST API 包装在 NodeJS thur HTTP 协议中。

Service Bus 中的 AMQP 支持详情，请参考https://azure.microsoft.com/en-us/documentation/articles/service-bus-amqp-overview/。

ServiceBus的REST API请参考https://msdn.microsoft.com/en-us/library/azure/hh780717.aspx。

AMQP 是一种二进制应用层协议，旨在高效地 支持各种消息传递应用程序和通信 模式。 - from WikiPedia

但是 HTTP 是一个文本协议。

消息格式如下，请参考工件http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#section-message-format的Message Format部分。并且AMQP规范可以参考http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-overview-v1.0-os.html。

                                                     Bare Message
                                                            |
                                      .---------------------+--------------------.
                                      |                                          |
 +--------+-------------+-------------+------------+--------------+--------------+--------+
 | header | delivery-   | message-    | properties | application- | application- | footer |
 |        | annotations | annotations |            | properties   | data         |        |
 +--------+-------------+-------------+------------+--------------+--------------+--------+
 |                                                                                        |
 '-------------------------------------------+--------------------------------------------'
                                             |
                                      Annotated Message

所以用Java发送的消息或者用NodeJS发送的消息被序列化成不同的结果。

AMQP的body内容中\uXXXX的内容是Unicode字符。

Unicode字符\u0006是Acknowledge控制字符，请参考https://en.wikipedia.org/wiki/Acknowledge_character了解。

而Unicode字符\u001a是替代控制字符，请参考https://en.wikipedia.org/wiki/Substitute_character。

它们限制了消息头中元数据的开始和结束。