AQ – We Do Streams

Oracle AQ Buffered Queues 101

This post illustrates some of the features of Oracle Advanced Queuing
Buffered Messages.
For more details, check Oracle® Streams Advanced Queuing User’s Guide,
DBMS_AQADM, DBMS_AQ, AQ Types or wait for some of the coming posts of
this blog ;-).
Obviously, we’ll be taking some shortcuts for now, but we’ll explore
basics of buffered messages, like:

1. Create a user, tablespace and grant the privileges to use Advanced
Queuing
2. Create a queue table, a multiple consumer queue and start it
3. Create subscribers for the queue
4. Create a procedure to enqueue buffered messages
5. Create a procedure to dequeue buffered messages
6. Enqueue & dequeue messages
7. Monitor queues and subscribers
8. Illustrate how you can loose messages by crashing an instance
9. Illustrate Flow Control and Spilled Message
10. Clean up the environment

1. Create a user, tablespace and grant the privileges

First of all, we’ll create a user, named AQDEMO, to own and use the
message queues.
In real life, you’ll probably separate the queue owner from message
publishers and consumers.
However, to keep that demo short, we’ll use AQDEMO to publish and
consume messages too.
It’s worth to notice that a tablespace that contains queue tables
doesn’t support Tablespace Point In Time Recovery (TSPITR).
For that reason and to ease the management of our AQ environment,
we’ll store the queues in a separate tablespace called AQ.
Last but not least if you want to share queues across multiple databases,
you should a use a distributed environment and set the global_names
parameter of all the databases to true.
The set of command to configure your database looks like below:

connect / as sysdba

alter system
set global_names=true;

select *
from global_name;

GLOBAL_NAME
———–
BLACK

create tablespace aq
datafile ‘/u01/app/oracle/oradata/BLACK/aq01.dbf’
size 25M autoextend on maxsize 266M;

create user aqdemo
identified by aqdemo
default tablespace aq
temporary tablespace temp;

grant connect, resource to aqdemo;
grant execute on dbms_aqadm to aqdemo;
grant execute on dbms_aq to aqdemo;

2. Create a queue table, a multiple consumer queue and start it

To perform these steps, we’ll connect as AQDEMO and will use:

dbms_aqadm.create_queue_table to create the queue table
dbms_aqadm.create_queue to create the queue
dbms_aqadm.start_queue to start the queue

To keep it simple, we’ll won’t use a user defined type for the message
but just a RAW:

connect aqdemo/aqdemo

begin
dbms_aqadm.create_queue_table(
queue_table =>’myqueue_table’,
queue_payload_type =>’RAW’,
storage_clause =>’tablespace aq’,
multiple_consumers => true,
comment =>’AQ Demo’,
compatible=>’11.1′);
end;
/

begin
dbms_aqadm.create_queue(
queue_name => ‘myqueue’,
queue_table => ‘myqueue_table’);
end;
/

begin
dbms_aqadm.start_queue(
queue_name => ‘myqueue’,
enqueue => true,
dequeue => true);
end;
/

Note:
If you plan to create a queue in another schema,
you can prefix the queue table or queue name with the schema name like
AQDEMO.MYQUEUE_TABLE.
If you want to mix cases, you can surround the name with double-quotes,
like “AQDEMO”.”myqueue_table” but that’s probably easier to use uppercase.

Once the queue table created, several tables and views are used to store
and query the associated data; here is an example with AQ$[QUEUE_TABLE]:

desc aq$myqueue_table

Name Null? Type
——————— ——– —————————-
QUEUE VARCHAR2(30)
MSG_ID NOT NULL RAW(16)
CORR_ID VARCHAR2(128)
MSG_PRIORITY NUMBER
MSG_STATE VARCHAR2(16)
DELAY DATE
DELAY_TIMESTAMP TIMESTAMP(6)
EXPIRATION NUMBER
ENQ_TIME DATE
ENQ_TIMESTAMP TIMESTAMP(6)
ENQ_USER_ID VARCHAR2(30)
ENQ_TXN_ID VARCHAR2(30)
DEQ_TIME DATE
DEQ_TIMESTAMP TIMESTAMP(9)
DEQ_USER_ID VARCHAR2(30)
DEQ_TXN_ID VARCHAR2(30)
RETRY_COUNT NUMBER
EXCEPTION_QUEUE_OWNER VARCHAR2(30)
EXCEPTION_QUEUE VARCHAR2(30)
USER_DATA BLOB
PROPAGATED_MSGID RAW(16)
SENDER_NAME VARCHAR2(30)
SENDER_ADDRESS VARCHAR2(1024)
SENDER_PROTOCOL NUMBER
ORIGINAL_MSGID RAW(16)
ORIGINAL_QUEUE_NAME VARCHAR2(30)
ORIGINAL_QUEUE_OWNER VARCHAR2(30)
EXPIRATION_REASON VARCHAR2(31)
CONSUMER_NAME VARCHAR2(30)
ADDRESS VARCHAR2(1024)
PROTOCOL NUMBER

3. Create subscribers for the queue

In the case of Publish/Subscribe, you have to create subscribers to
dequeue messages.
In our example, we’ll register 2 subscribers called SUBSCRIBER1 and
SUBSCRIBER2 for that purpose:

connect aqdemo/aqdemo

declare
subscriber sys.aq$_agent;
begin
subscriber := sys.aq$_agent(
‘subscriber1’,
‘myqueue’,
null);
dbms_aqadm .add_subscriber(
queue_name => ‘myqueue’,
subscriber => subscriber,
delivery_mode => dbms_aqadm.buffered);
end;
/

declare
subscriber sys.aq$_agent;
begin
subscriber := sys.aq$_agent(
‘subscriber2’,
‘myqueue’,
null);
dbms_aqadm .add_subscriber(
queue_name => ‘myqueue’,
subscriber => subscriber,
delivery_mode => dbms_aqadm.buffered);
end;
/

You can check queue subscribers by querying the AQ$MYQUEUE_TABLE_S
view (or AQ$_MYQUEUE_TABLE_S table) like below:

col name format a12
col address format a40

select name, address
from aq$myqueue_table_s
where queue=’MYQUEUE’;

NAME ADDRESS
———— ————————-
SUBSCRIBER1
SUBSCRIBER2

4. Create a procedure to enqueue buffered messages

The procedure below can be used to enqueue buffered messages;
The options and properties used to enqueue the messages are the following:

The message is a RAW(10) passed from the procedure parameter as an
hexadecimal string

The list of message recipients are SUBSCRIBER1 and SUBSCRIBER2
The message is a buffered message (dbms_aq.buffered)

connect aqdemo/aqdemo

create or replace procedure demo_enqueue(p_hexamsg varcha r2) is
enqueue_options DBMS_AQ.enqueue_options_t;
message_properties DBMS_AQ.message_properties_t;
recipients DBMS_AQ.aq$_recipient_list_t;
message_handle RAW(16);
message RAW(10);
begin
message := hextoraw(p_hexamsg);
recipients(1) := sys.aq$_agent(‘SUBSCRIBER1’, ‘MYQUEUE’, NU LL);
recipients(2) := sys.aq$_agent(‘SUBSCRIBER2’, ‘MYQUEUE’, NU LL);
message_properties.recipient_list := recipients;
enqueue_options.visibility := dbms_aq.immediate;
enqueue_options.delivery_mode := dbms_aq.buffered;
dbms_aq.enqueue(
queue_name => ‘MYQUEUE’,
enqueue_options => enqueue_options,
message_properties => message_properties,
payload => message,
msgid => message_handle);
commit;
end;
/

Once the procedure created, you can easily enqueue messages from the
AQDEMO user:

exec demo_enqueue(‘00000000000000000001’);

5. Create a procedure to dequeue buffered messages

The procedure below dequeues the buffered messages and display it with
DBMS_OUTPUT.PUT_LINE;
it takes the subscriber name as a parameter

connect aqdemo/aqdemo

set serveroutput on

create or replace procedure demo_dequeue(p_subscriber varchar2)
is
dequeue_options dbms_aq.dequeue_options_t;
message_properties dbms_aq.message_properties_t;
message_handle RAW(16);
message RAW(10);
no_messages exception;
pragma exception_init(no_messages, -25228);
begin
dequeue_options.wait := dbms_aq.no_wait;
dequeue_options.consumer_name := p_subscriber;
dequeue_options.navigation := dbms_aq.first_message;
dequeue_options.visibility := dbms_aq.immediate;
dequeue_options.delivery_mode := dbms_aq.buffered;
loop
begin
dbms_aq.dequeue(
queue_name => ‘myqueue’,
dequeue_options => dequeue_options,
message_properties => message_properties,
payload => message,
msgid => message_handle);
dbms_output.put_line(‘Message: ‘|| hextoraw(message) );
dequeue_options.navigation := dbms_aq.next_message;
end;
end loop;
exception
when no_messages then
dbms_output.put_line(‘No more messages’);
commit;
end;
/

You can dequeue messages with demo_dequeue like below:

set serveroutput on
exec demo_dequeue(‘SUBSCRIBER1’);

Message: 00000000000000000001
No more messages

6. Enqueue & Dequeue messages

Here is a sequence of enqueue and dequeue of buffered messages:

exec demo_enqueue(‘00000000000000000002’);
exec demo_enqueue(‘00000000000000000003’);

set serveroutput on
exec demo_dequeue(‘SUBSCRIBER1’);

Message: 00000000000000000002
Message: 00000000000000000003
No more messages

exec demo_enqueue(‘00000000000000000004’);

exec demo_dequeue(‘SUBSCRIBER1’);
Message: 00000000000000000004
No more messages

set lines 100
col subscriber_name format a15
col subscriber_type format a15
col startup_time format a18
col total_dequeued_msg format 999,999

select queue_id,
subscriber_name,
subscriber_type,
startup_time,
total_dequeued_msg
from v$buffered_subscribers
where queue_schema=’AQDEMO’
and queue_name=’MYQUEUE’;

QUEUE_ID SUBSCRIBER_NAME SUBSCRIBER_TYPE STARTUP_TIME
TOTAL_DEQUEUED_MSG
——– ————— ————— ——————
——————
73821 SUBSCRIBER1 SUBSCRIBER 03-JAN-09 17:55:22 0
73821 SUBSCRIBER2 SUBSCRIBER 03-JAN-09 17:55:22 0

7. Monitor queues and subscribers

You can query the V$BUFFERED_QUEUE and V$BUFFERED_SUBSCRIBERS fixed views
to check the number of messages in the queues and how many messages
have been dequeued:

connect / as sysdba

alter session set nls_date_format=’DD-MON-YY HH24:MI:SS’;

col queue_id format 999999
col startup_time format a18
col num_msgs format 999,999
col spill_msgs format 999,999
set lines 100

select queue_id
, startup_time
, sysdate
, num_msgs
, spill_msgs
from v$buffered_queues
where queue_schema=’AQDEMO’
and queue_name=’MYQUEUE’;

QUEUE_ID STARTUP_TIME SYSDATE NUM_MSGS SPILL_MSGS
——– —————— —————— ——– ———-
74026 04-JAN-09 00:13:12 04-JAN-09 00:17:17 4 0

set lines 100
col subscriber_name format a15
col subscriber_type format a15
col startup_time format a18
col total_dequeued_msg format 999,999

select queue_id
, subscriber_name
, subscriber_type
, startup_time
, total_dequeued_msg
from v$buffered_subscribers
where queue_schema=’AQDEMO’
and queue_name=’MYQUEUE’;

QUEUE_ID SUBSCRIBER_NAME SUBSCRIBER_TYPE STARTUP_TIME
TOTAL_DEQUEUED_MSG
——– ————— ————— ——————
——————
74026 SUBSCRIBER1 SUBSCRIBER 04-JAN-09 00:13:12 4
74026 SUBSCRIBER2 SUBSCRIBER 04-JAN-09 00:13:12 0

You can dequeue the message from the other subscriber:

exec demo_dequeue(‘SUBSCRIBER2’);

Message: 00000000000000000001
Message: 00000000000000000002
Message: 00000000000000000003
Message: 00000000000000000004
No more messages

8. Illustrate how you can loose messages by crashing an instance

You can easily show that a buffered messages can be lost.
Here is a simple scenario to illustrate that point with a bounce of
the Oracle instance:

connect aqdemo/aqdemo
exec demo_enqueue(‘00000000000000000005’);

connect / as sysdba
startup force

connect aqdemo/aqdemo
exec demo_enqueue(‘00000000000000000006’);

set serveroutput on
exec demo_dequeue(‘SUBSCRIBER1’);

Message: 00000000000000000006
No more messages

exec demo_dequeue(‘SUBSCRIBER2’)
Message: 00000000000000000006
No more messages

As you can see above, the message 00000000000000000005 has never been
dequeued but is lost;
The same scenario illustrates, as expected,
that statistics of queue and subscribers usage are reset when an
instance is bounced:

connect / as sysdba

alter session set nls_date_format=’DD-MON-YY HH24:MI:SS’;

col queue_id format 999999
col startup_time format a18
col num_msgs format 999,999
col spill_msgs format 999,999
set lines 100

select queue_id
, startup_time
, sysdate
, num_msgs
, spill_msgs
from v$buffered_queues
where queue_schema=’AQDEMO’
and queue_name=’MYQUEUE’;

QUEUE_ID STARTUP_TIME SYSDATE NUM_MSGS SPILL_MSGS
——– —————— —————— ——– ———-
74026 04-JAN-09 00:30:03 04-JAN-09 00:33:55 0 0

set lines 100
col subscriber_name format a15
col subscriber_type format a15
col startup_time format a18
col total_dequeued_msg format 999,999

select queue_id
, subscriber_name
, subscriber_type
, startup_time
, total_dequeued_msg
from v$buffered_subscribers
where queue_schema=’AQDEMO’
and queue_name=’MYQUEUE’;

QUEUE_ID SUBSCRIBER_NAME SUBSCRIBER_TYPE STARTUP_TIME
TOTAL_DEQUEUED_MSG
——– ————— ————— ——————
——————
74026 SUBSCRIBER1 SUBSCRIBER 04-JAN-09 00:30:03 1
74026 SUBSCRIBER2 SUBSCRIBER 04-JAN-09 00:30:03 1

9. Illustrate Flow Control and Spilled Messages

One of the benefit of using buffered messages is that they remain in
memory and are not stored on disks.
A direct consequence of that is that buffered messages are not logged
and as we’ve seen previously can be lost.
The devil being in the details, the memory allocated for the Streams
Pool, that store the messages,
is not unlimited and compete with other memory pools.
To prevent that memory to increase without limit Oracle offers several
mechanisms;
One is Publisher Flow Control that prevents messages from being
enqueued when the messages are not dequeued fast enough.
Another one is the ability for messages to spill on disks in the queue table.
Here is a simple test case shows these features :

connect aqdemo/aqdemo

begin
for i in 7..100000 loop
demo_enqueue(lpad(to_char(i),20,’0′));
end loop;
end;
/

begin
*
ERROR at l ine 01:00
ORA-25307: En queue rate too high, flow co ntrol enabled
ORA-06512: at “SYS.DBMS_AQ”, line 6
ORA-06512: at “SYS.DBMS_AQ”, line 216
ORA-06512: at “AQDEMO.DEMO_ENQUEU E”, l ine 14
ORA-06512: at line 3

If you monitor the content of the queues right after you’ve enqueued
them, you’ll see that messages are in the memory associated with the
queue:

connect / as sysdba

select queue_id
, startup_time
, sysdate
, num_msgs
, spill_msgs
from v$buffered_queues
where queue_schema=’AQDEMO’
and queue_name=’MYQUEUE’;

QUEUE_ID STARTUP_TIME SYSDATE NUM_MSGS SPILL_MSGS
——– —————— ——— ——– ———-
74026 04-JAN-09 04-JAN-09 4,021 0

select MSG_STATE
, count(*)
from aqdemo.aq$myqueue_table
where queue=’MYQUEUE’
group by MSG_STATE;

MSG_STATE COUNT(*)
——— ——–
IN MEMORY 8042

col bytes format 999,999,999
col name format a30
set pages 1000

select name, bytes
from v$sgastat
where pool=’streams pool’
order by name;

NAME BYTES
————————– ————
Sender info 8,484
deqtree_kgqmctx 80
fixed allocation callback 260
free memory 5,702,748
image handles 337,452
kgqbt_alloc_block 99,456
kgqmdm_fl_2 241,288
kgqmsub 144
kodpaih3 image 8,130,416
kwqbcqini:spilledovermsgs 1,968
kwqbdaspl:spilledovermsgs 172,896
kwqbsinfy:bms 305,368
kwqbsinfy:bqg 808
kwqbsinfy:mpr 1,928,180
kwqbsinfy:sta 208
msgtree_kgqmctx 80
name_kgqmsub 32
recov_kgqbtctx 8,192
recov_kgqmctx 616
recov_kgqmsub 336
spilled:kwqbl 2,316
spilled:kwqbm 8,624
substree_kgqmctx 80
time manager index 80

And you’ll also see the publisher is “IN FLOW CONTROL”:

select publisher_state
from v$buffered_publishers
where queue_schema=’AQDEMO’
and queue_name=’MYQUEUE’;

PUBLISHER_STATE
———————————————————–
IN FLOW CONTROL: INSUFFICIENT MEMORY AND UNBROWSED MESSAGES

If you wait for a few minutes, you’ll see messages start to spill on disks:

select queue_id
, startup_time
, sysdate
, num_msgs
, spill_msgs
from v$buffered_queues
where queue_schema=’AQDEMO’
and queue_name=’MYQUEUE’;

QUEUE_ID STARTUP_TIME SYSDATE NUM_MSGS SPILL_MSGS
——– —————— ——— ——– ———-
74026 04-JAN-09 04-JAN-09 4,021 4,021

select MSG_STATE, count(*)
from aqdemo.aq$myqueue_table
where queue=’MYQUEUE’
group by MSG_STATE;

MSG_STATE COUNT(*)
——— ——–
SPILLED 8042

You can dequeue the messages as you’ve done already

connect aqdemo/aqdemo

begin
for i in 1..5000 loop
demo_dequeue(‘SUBSCRIBER1’);
demo_dequeue(‘SUBSCRIBER2’);
end loop;
end;
/

10. Clean up the environment

We can conclude this short demonstration of buffered messages by a bit
of clean up.
I’ll dig into more details in the coming posts

connect / as sysdba

drop procedure aqdemo.demo_enqueue;
drop procedure aqdemo.demo_dequeue;

declare
subscriber sys.aq$_agent;
begin
subscriber := sys.aq$_agent(‘subscriber1’,
‘aqdemo.myqueue’,
null);
dbms_aqadm.remove_subscriber(
queue_name => ‘aqdemo.myqueue’,
subscriber => subscriber);
subscriber := sys.aq$_agent(‘subscriber2’,
‘aqdemo.myqueue’,
null);
dbms_aqadm.remove_subscriber(
queue_name => ‘aqdemo.myqueue’,
subscriber => subscriber);
end;
/

begin
dbms_aqadm.stop_queue(
queue_name => ‘aqdemo.myqueue’,
enqueue => true,
dequeue => true);
end;
/

begin
dbms_aqadm.drop_queue(
queue_name => ‘aqdemo.myqueue’);
end;
/

begin
dbms_aqadm.drop_queue_table(
queue_table =>’aqdemo.myqueue_table’);
end;
/

drop user aqdemo;

drop tablespace aq
including contents and datafiles;

Advertisements