dequeue sélective de messages sans rapport avec Oracle Advanced Queuing

Question

Cette question fait référence à la dequeueing des messages dans Oracle Streams Advanced Queuing .

Je dois veiller à ce que les messages qui sont liés les uns aux autres sont traitées de manière séquentielle.

Par exemple, supposons que la file d'attente est ensemencé avec les quatre messages qui ont une référence de transaction appelée domaine lié d'affaires (txn_ref) et deux des messages (1,3) appartiennent à la même transaction (000001):

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

Supposons aussi que je suis en cours d'exécution 4 threads / processus qui souhaitent dequeue de cette file d'attente. Les points suivants doivent se produire:

1. 1 fil dequeues Message n ° 1
2. 2 fils dequeues Message n ° 2
3. fil 3 dequeues Message n ° 4 (parce que le message n ° 3 est lié à # 1 et # 1 n'a pas encore terminé).
4. fil 4 blocs en attente d'un message
5. 1 fil engage son travail pour le message # 1
6. fil 4 (ou peut-être le fil 1) dequeues un message # 3.

Ma première pensée était que je pouvais y parvenir avec une condition dequeue où le ENQ_TIME (heure enqueue) est plus tard que tout autre ENQ_TIME de tous les messages qui ont le même TXN_REF. Mais mon problème est de savoir comment référencer le TXN_REF d'un message que je ne l'ai pas encore sélectionné, afin de le sélectionner. par exemple.

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

Est-il possible d'obtenir ce que je veux ici?

Answer 1

Pour répondre directement à votre question, cela peut être réalisé en utilisant le champ correlation (appelé CORRID dans le tableau), qui est conçue à cet effet.

Alors, sur le enqueue, vous devriez utiliser la méthode AQMessageProperties.setCorrelation() avec la valeur TXN_REF comme paramètre. Ensuite, dans votre état, vous feriez quelque chose comme ceci:

// 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);

Answer 2

Une stratégie que vous pouvez essayer, si possible, utilise des groupes de message. Oracle Documentation décrit brièvement, mais je trouve cet article Crapaud du monde être beaucoup plus utile. Fondamentalement, vous configurez la table de file d'attente pour traiter tous les messages commis en même temps que l'un « groupe ». Lorsque dequeueing, un seul utilisateur à la fois peut dequeue d'un « groupe » de messages.