Websphere MQ，接收大 (100 MB) 消息

Question

我正在尝试从 Websphere MQ 队列中读取一个 100 MB 的文件。文件应包含大约。 800.000 条记录。但是在某个行号（通常在 90.000 左右）之后，我只得到空的（或者可能是 �，这取决于我尝试查看文件的方式）。

我被告知队列应该允许 100 MB 消息。

我的代码：

            var mqQueueManager = getMqQueueManager(ConfigurationManager.AppSettings["MQ_QueueManager"]);
            var mqQueue = mqQueueManager.AccessQueue(mqReceiveQueueName,
                                                        MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_FAIL_IF_QUIESCING);

            MQGetMessageOptions gmo = new MQGetMessageOptions();
            gmo.Options = MQC.MQGMO_NO_WAIT;

            for (int i = 0; i < maxMessagesToRead; i++)
            {
                var mqMessage = new MQMessage();
                mqQueue.Get(mqMessage, gmo);

                messageTexts.Add(mqMessage.ReadString(mqMessage.MessageLength));
            }

这段代码有问题吗？还是发送方的错误？

Answer 1

不清楚要问什么，我相信这是问题的根源。虽然 MQ 确实可以处理最大 100MB 的消息，但这并不是问题的意思。

我正在尝试从 Websphere MQ 队列中读取一个 100 MB 的文件。文件 应该包含大约。 800.000 条记录。但是在某个行号之后 （通常在 90.000 左右）

隐藏在单个语句中的是对以下内容的引用：

• 文件
• 队列
• 记录
• 行

我确定您询问的每个人都在响应队列，因此反复保证 MQ 可以处理 100MB 消息。但代码似乎正在重新组装一个 文件，该文件已被分成一个 record/row 每个 message。这是一个完全不同的用例。

一次拿这些，在 MQ 中来回传递 100MB 消息是完全可能的。这样做需要将沿该路径的每个 QMgr、通道和队列从 MAXMSGL 的 4MB 默认值进行配置。不太明显的是，如果使用线性日志和/或消息在同步点下处理，则必须有足够的日志空间来包含多个 100MB 事务。默认的日志范围计数和大小预计不会有 100MB 消息。最后，应用程序需要提供足够大的缓冲区，并确保不要告诉 MQ 允许截断消息。

这种通过 MQ 移动文件的方法仅限于 100MB 以下的文件，但具有原子操作的明显优势 - 您可以 GET/PUT 整个文件或根本不获取。没有遗漏的部分，没有乱序等。这大大简化了代码。

第二种可能性是代码正在重新组装一个已拆分为许多消息的文件。似乎令人怀疑的是，发布的代码 sn-p 实际上想要将多条 100MB 消息相互附加，所以我假设它正在尝试从许多消息中重新组装一个文件。 （我相信最多 800,000 条。）在这种情况下，问题在于单个工作单元中可以包含多少条消息。代码 sn-p 省略了任何有意义的错误处理，所以我假设同步点处理也被省略了。

重组文件时出现的错误包括乱序消息和丢失消息。即使是最简单的基于 MQ 的文件移动器（每个文件使用多条消息）也需要使用同步点来确保整个文件作为一个单元发送或接收。但是 MQ 预计单个工作单元中不会有 800,000 条消息，因此您需要调整日志范围 和 MAXUMSGS .

当然，移动文件并在将文件拆分为多条消息时保证其完整性并非易事，当考虑到所有排列时，结果看起来很像 MQ 托管文件传输。

回到问题上来，如果文件是在单个消息中传输的，并且没有设置截断消息的 API 选项，那么 MQ API 中没有任何内容会影响之后的消息一些行数或消息长度。在这种情况下，解释垃圾超过某个点的唯一方法是发送应用程序发送了它。 （例如，分配 100MB 缓冲区，填满 50MB，然后发送 100MB 消息。）

但是，如果文件被拆分为从代码中显示的许多消息，则必须执行上述调整。这包括将MAXUMSGS 增加到大于800,000 (doc link) 的值，并确保主日志区和辅助日志区有足够的空间来容纳一些大型工作单元。详情请见Calculating the size of the log。

顺便说一句，您可能还希望每次都重置MQGMO，因为正如该对象的Overview 中所述，它既是输入 也是 输出。

修改后的代码看起来更像这样：

            MQGetMessageOptions gmo = new MQGetMessageOptions();

            for (int i = 0; i < maxMessagesToRead; i++)
            {
                var mqMessage = new MQMessage();
                gmo.Options = MQC.MQGMO_NO_WAIT;
                mqQueue.Get(mqMessage, gmo);