On a normal queue, each message goes to one consumer. A multiple-consumer queue works like publish and subscribe: every subscriber gets its own copy of each message, and a rule can limit a subscriber to the messages it cares about. DBMS_AQADM.ADD_SUBSCRIBER registers the subscribers.
Code for This Guide
The main examples are in the examples/pkg-messaging folder of the Oracle Database 26ai code repository on GitHub, each with its output. They use NIMBUS, the sample schema of a fictional airline, which you can install with the scripts in the setup/nimbus folder.
They come from Oracle Database 26ai SQL and PL/SQL Book.
Syntax
dbms_aqadm.create_queue_table(queue_table, payload_type, multiple_consumers => true);
dbms_aqadm.add_subscriber(queue_name, sys.aq$_agent('NAME', null, null),
rule => 'condition on tab.user_data');
-- dequeue as a subscriber
v_options.consumer_name := 'NAME';The example uses the GATE_CHANGE_T type from How to Create Queues with DBMS_AQADM.
Two Subscribers, One with a Rule
Example:
begin
dbms_aqadm.create_queue_table('ALERT_QT', 'GATE_CHANGE_T', multiple_consumers => true);
dbms_aqadm.create_queue('ALERT_Q', 'ALERT_QT');
dbms_aqadm.start_queue('ALERT_Q');
dbms_aqadm.add_subscriber('ALERT_Q', sys.aq$_agent('CREW_APP', null, null));
dbms_aqadm.add_subscriber('ALERT_Q', sys.aq$_agent('PAX_APP', null, null),
rule => q'[tab.user_data.flight_no like 'NM1%']');
end;
/
declare
v_enq dbms_aq.enqueue_options_t;
v_deq dbms_aq.dequeue_options_t;
v_props dbms_aq.message_properties_t;
v_msg gate_change_t;
v_msgid raw(16);
begin
dbms_aq.enqueue('ALERT_Q', v_enq, v_props, gate_change_t('NM150', 'A1', 'B12'), v_msgid);
dbms_aq.enqueue('ALERT_Q', v_enq, v_props, gate_change_t('NM205', 'D2', 'D4'), v_msgid);
commit;
v_deq.wait := dbms_aq.no_wait;
for c in (select column_value as consumer
from sys.odcivarchar2list('CREW_APP', 'PAX_APP')) loop
v_deq.consumer_name := c.consumer;
v_deq.navigation := dbms_aq.first_message; -- start again for each consumer
begin
loop
dbms_aq.dequeue('ALERT_Q', v_deq, v_props, v_msg, v_msgid);
dbms_output.put_line(rpad(c.consumer, 10) || v_msg.flight_no || ' -> '
|| v_msg.new_gate);
end loop;
exception
when others then if sqlcode != -25228 then raise; end if;
end;
end loop;
commit;
end;
/Output:
PL/SQL procedure successfully completed. CREW_APP NM150 -> B12 CREW_APP NM205 -> D4 PAX_APP NM150 -> B12 PL/SQL procedure successfully completed.
- CREW_APP subscribes to everything, so it receives both gate changes.
- PAX_APP has a rule on the payload, flight numbers starting with NM1, so it receives only NM150.
- Each consumer dequeues with its own CONSUMER_NAME and starts from FIRST_MESSAGE.
A message stays in the queue until every subscriber it was meant for has dequeued it.
Things to Know
- Subscribers added later receive only messages enqueued after they subscribed.
- USER_QUEUE_SUBSCRIBERS lists the subscribers and their rules; REMOVE_SUBSCRIBER deletes one.
- A message can also name its own recipients in the RECIPIENT_LIST property, overriding the subscriber list.
Cleanup:
exec dbms_aqadm.drop_queue_table('ALERT_QT', force => true)Related Guides
Conclusion
Multiple-consumer queues deliver each message to every subscriber, and subscriber rules filter what each one receives. Add subscribers with ADD_SUBSCRIBER and dequeue with each subscriber's CONSUMER_NAME.
See my articles more often when you search Google.
