Oracle Advanced Queuing 中不相关消息的选择性出队答案

【问题标题】：Selective dequeue of unrelated messages in Oracle Advanced QueueingOracle Advanced Queuing 中不相关消息的选择性出队
【发布时间】：2011-04-27 19:59:18
【问题描述】：

本题涉及 Oracle Streams 高级队列 中的消息出列。

我需要确保相互关联的消息按顺序处理。

例如，假设队列中包含四个消息，这些消息具有称为事务引用 (txn_ref) 的业务相关字段，并且其中两个消息 (1,3) 属于同一事务 (000001)：

id | txn_ref | 
---+---------+
 1 | 000001  |
 2 | 000002  |
 3 | 000001  |
 4 | 000003  |

还假设我正在运行 4 个希望从该队列中出列的线程/进程。应该会发生以下情况：

线程 1 将消息 #1 出列
线程 2 将消息 #2 出列
线程 3 将消息 #4 出列（因为消息 #3 与 #1 相关，而 #1 尚未完成）。
线程 4 块等待消息
线程 1 提交消息 #1 的工作
线程 4（或者线程 1）将消息 #3 出列。

我最初的想法是，我可以通过出列条件来实现这一点，其中 ENQ_TIME（入队时间）不晚于具有相同 TXN_REF 的所有消息的任何其他 ENQ_TIME。但我的问题是如何引用我尚未选择的消息的 TXN_REF 以便选择它。例如

// Java API
String condition = "ENQ_TIME = (select min(ENQ_TIME) from AQ_TABLE1 where ??";
dequeueOption.setCondition(condition);

这里可以实现我想要的吗？

【问题讨论】：

标签： oracle advanced-queuing oracle-aq oracle-streams

【解决方案1】：

要回答您的直接问题，可以使用为此目的设计的correlation 字段（在表中称为CORRID）来实现。

因此，在入队时，您将使用AQMessageProperties.setCorrelation() 方法并将 TXN_REF 值作为参数。然后，在你的情况下，你会做这样的事情：

// Java API
String condition = "tab.ENQ_TIME = (select min(AQ_TABLE1.ENQ_TIME) from AQ_TABLE1 self where tab.CORRID=AQ_TABLE1.CORRID)";
dequeueOption.setCondition(condition);

【讨论】：

我想要做的是将消息出列，除非相关 id 与当前正在由不同工作人员处理的消息匹配。如果我可以识别当前正在处理的相关 ID，那么我可以将条件设置为 corrid not in {a,b ... z}。
那么您可能需要两个队列：一个用于上述消息本身，另一个用作“标头”消息（每个 TXN_REF 1 个）。一个线程首先将标头出列以获取 TXN_REF/CORRID，然后该线程将处理具有该 CORRID 的所有记录。
有趣的想法。头消息队列就像一个令牌系统。入队进程应该仅在队列中不存在头消息时将其入队。（这可能吗，我想知道）。然后，因为我希望每个事务处理不超过 1 条消息，所以出队过程可以使用标头从消息队列中选择性地出队 single 消息。当它处理完该消息后，如果有另一条相关消息要处理，它会将标头消息放回队列中。复杂，但有一定的希望。
附言。我非常感谢您的帮助。它当然值得赏金。

【解决方案2】：

如果可能的话，您可以尝试的一种策略是使用消息组。 Oracle Documentation 简要描述了它，但我发现this Toad World article 更有用。基本上，您设置队列表以将同时提交的所有消息视为一个“组”。出列时，一次只有一个用户可以从“组”消息中出列。

【讨论】：

这是有希望的，但还不够。我的消息在逻辑上是相关的，但它们没有在同一个事务中排队。我还能使用消息组吗？
我几乎可以肯定他们必须是同一交易的一部分。