.net XMS 中的 IBM-MQ 阅读器，用于一一确认已处理的消息答案

【问题标题】：IBM-MQ reader in .net XMS to ack processed messages one-by-one.net XMS 中的 IBM-MQ 阅读器，用于一一确认已处理的消息
【发布时间】：2011-11-30 13:42:53
【问题描述】：

我正在实现一个组件，当它们可用时从特定队列中读取所有消息，但应该只在消息内容被处理和持久化后从队列中异步删除消息。我们读取消息的速度比我们确认消息的速度快（例如，我们可以在准备好确认第一条消息之前读取已读取 10 条消息）。当前实现使用 XMS API，但如果 XMS 不适合这些用途，我们可以切换到 MQI。

我们尝试了两种方法来尝试解决这个问题，但它们都有一些缺点，使它们无法接受。我希望有人能提出更好的方法。

第一个实现在专用线程中使用IMessageConsumer 来读取所有消息并在它们可用时发布它们的内容。处理完消息后，将调用message.Acknowledge() 方法。会话是使用AcknowledgeMode.ClientAcknowledge 创建的。这种方法的问题在于，根据文档，这会确认（并删除）所有已收到的未确认消息。对于上面的示例，这意味着所有 10 条已读消息都将在第一次调用时得到确认。所以这并不能真正解决问题。由于我们需要的读取吞吐量，我们无法真正修改此解决方案以在读取第二个消息之前等待第一个消息的确认，等等。

第二个实现在确定的线程中使用IQueueBrowser 来读取所有消息并发布其内容。这不会在读取时从队列中删除消息。然后，一个单独的专用线程（在 BlockingQueue 上）等待已处理消息的 JMS 消息 ID。对于其中的每一个，它会构造一个专用的IMessageConsumer（使用带有 JMSMessageID 的消息选择器）来读取消息并确认它。（IQueueBrowser 与专用 IMessageConsumer 的配对是 XMS 文档中关于 队列浏览器 的部分推荐的。）这种方法确实按预期工作，但正如人们想象的那样，它太 CPU-在 MQ 服务器上密集。

【问题讨论】：

是的，就是这样，消息不能被随机确认。一个消息确认将确认所有较早的消息。 MQ .NET 中也没有选项。根据您的描述，您似乎可以通过一次 ack 调用确认所有消息，但您更关心吞吐量。您期望的吞吐量是多少？
预期吞吐量：最活跃队列每天最多 100 万条消息，MQ 服务器上所有队列每天最多 1000 万条消息。
澄清一下：我们不能真正顺序读取和处理每条消息（在读取下一条之前）的主要原因是实际处理有相当长的前置时间 - 这就是我们需要的原因同时处理多条消息。
您提出的两种方法似乎都依赖于应用程序的单个实例。使用多个应用程序实例、事务会话和 COMMIT 有什么问题？性能报告都显示，使用多个应用程序实例可以最大限度地提高吞吐量，而水平扩展是您的场景中最常用的方法之一。
@T.Rob 您将如何构建它？多个线程，每个线程都有自己的 Session 和 Consumer；每次阅读一条消息然后等待它被处理？另外：你会把你的评论变成答案吗:-)

标签： c# .net ibm-mq

【解决方案1】：

问题中提出的两种方法似乎都依赖于应用的单个实例。使用多个应用程序实例、事务会话和 COMMIT 有什么问题？性能报告（这些是名称类似于 MP** 的 SupportPacs）都显示了多个应用实例的吞吐量最大化，并且水平扩展是您的方案中最常用的方法之一。

为此的设计可以是多个应用程序实例，也可以是同一应用程序中的多个线程。使其正常工作的关键是要记住事务的范围是连接句柄。这意味着多线程应用程序必须为每个连接实例分派一个单独的线程，并且在同一个线程中读取消息。

处理流程是，应用程序使用事务处理会话对队列执行正常的MQGet，根据需要处理消息内容，然后发出MQCommit。（我将在示例中使用 MQ 本机 API 名称，因为这与语言无关。）如果这是一个 XA 事务，应用程序将调用 MQBegin 来启动事务，但对于单阶段提交，假定事务。在这两种情况下，MQCommit 都会结束从队列中删除消息的事务。当消息处于同步点时，没有其他应用程序实例可以检索它们； MQ 只是传递下一条可用消息。如果事务回滚，则任何线程的下一个MQGet 都会检索它，假设 FIFO 交付。

有一些样本在：
[WMQ install home]\tools\dotnet\samples\cs\xms\simple\wmq\
...而SimpleXAConsumer.cs 是一个示例，它显示了它的 XA 版本。非 XA 版本更简单，因为您不需要外部协调器、MQBegin 动词等。如果您从其中之一开始，请确保它们没有指定队列的独占使用，并且您可以使用相同的配置启动多个实例。或者，获取示例中包括创建连接、消息处理、连接关闭和销毁的部分，并将所有这些包装在线程生成器类中。

[在此处插入有关使用最新版本课程的常规建议。]

【讨论】：

所以我已经尝试过了，它按照描述的方式工作。但我确实有一个后续问题：我想在此设置中没有简单的方法可以跳过特定消息（在我的应用程序无法处理消息的情况下）并将其留在队列中以供另一个应用程序或未来的运行？
接受的模式是使用返工队列或异常队列。 IBM JMS 类将查找队列属性BOQNAME 和BOQTHRESH。一旦消息的回滚计数超过BOQTHRESH，消息将自动重新排队到BOQNAME 队列。这不需要额外的程序逻辑，只需定义队列，设置BOQTHRESH 并将BOQNAME 指向异常队列。一些应用程序遵循类似的模式，但添加了自己的逻辑来处理重新排队，这是一种更便携的方法。然后根据需要处理异常队列。