begin
  dbms_aqadm.stop_queue(queue_name => 'XXTEST_QUEUE', dequeue => true);
end;

/*создать тип*/
create Type message_typ as object ( 
  subject Varchar2(30), text Varchar2(80)
);

/*создать таблицу для очереди, очередь, запустить очередь*/
begin
  dbms_aqadm.create_queue_table( 
    queue_table        => 'XXTEST_QUEUE_TAB',
    queue_payload_type => 'MESSAGE_TYP',
    multiple_consumers => true);

  dbms_aqadm.create_queue(
    queue_name  => 'XXTEST_QUEUE',
    queue_table => 'XXTEST_QUEUE_TAB'); 

  dbms_aqadm.start_queue(
    queue_name => 'XXTEST_QUEUE');
end;

/*создать процедуру для добавления в очередь*/
create or replace procedure enqueue_msg( p_msg in varchar2 )
as
  enqueue_options    dbms_aq.enqueue_options_t;
  message_properties dbms_aq.message_properties_t;
  message_handle     RAW(16);
  message            message_typ;
begin

   message := message_typ('NORMAL MESSAGE',  p_msg );

   dbms_aq.enqueue(queue_name => 'XXTEST_QUEUE',
                   enqueue_options    => enqueue_options,
                   message_properties => message_properties,
                   payload            => message,
                   msgid              => message_handle);
end;

/*создать талицу, куда будут писаться сообщения из очереди*/
create table message_table( 
   msg           varchar2(4000)
  ,creation_date date
  ,info          varchar2(150)
);

/*создать процедуру для обработки сообщения из очереди*/
create or replace procedure XXTEST_CALLBACK( context raw,
                                      reginfo sys.aq$_reg_info,
                                      descr sys.aq$_descriptor,
                                      payload raw,
                                      payloadl number)
as
  dequeue_options    dbms_aq.dequeue_options_t;
  message_properties dbms_aq.message_properties_t;
  message_handle     raw(16);
  message            message_typ;
begin
  xxfnd_log_pkg.log(i_api_name => 'XXTEST_CALLBACK', i_msg => 'start', i_at => true);
  
  dequeue_options.msgid         := descr.msg_id;
  dequeue_options.consumer_name := descr.consumer_name;

  DBMS_AQ.DEQUEUE(
    queue_name         => descr.queue_name,
    dequeue_options    => dequeue_options,
    message_properties => message_properties,
    payload            => message,
    msgid              => message_handle);

  insert into message_table(msg, creation_date, info)
  values('Dequeued and processed "' || message.text || '"', sysdate, sys_context('USERENV', 'SID'));
  
  dbms_lock.sleep(1); --!!!

  commit;
exception
  when others then
    xxfnd_error.eh('XXTEST_CALLBACK');
end XXTEST_CALLBACK;

/*создать подписчика*/
begin
  dbms_aqadm.add_subscriber( 
    queue_name => 'XXTEST_QUEUE',
    subscriber =>  sys.aq$_agent('XXSUBSCRIBER', null, null));
end;

/*Регистрация листенера*/
DECLARE
 reginfo1     sys.aq$_reg_info;
 reginfolist  sys.aq$_reg_info_list;
BEGIN
 reginfo1 := sys.aq$_reg_info('INFONEXT.XXTEST_QUEUE:XXSUBSCRIBER',
                    DBMS_AQ.NAMESPACE_AQ,
                    'plsql://XXTEST_CALLBACK',
                    HEXTORAW('FF'));
 
 -- Create the registration info list
 reginfolist := sys.aq$_reg_info_list(reginfo1);
 
 -- do the registration
 sys.dbms_aq.register(reginfolist, 1);
END;

/*добавление множества сообщений*/
begin
  for r in 1..1000 loop
    enqueue_msg('test'||to_char(r));
  end loop;
end;

commit;

begin
  dbms_aqadm.stop_queue(queue_name => 'XXTEST_QUEUE', enqueue => false, dequeue => true);
end;

select * from all_queues q
where q.name = 'XXTEST_QUEUE';

begin
  dbms_aqadm.start_queue(queue_name => 'XXTEST_QUEUE');
end;