Kolejka Oracle Advance - brak działania w kolejce

Question

Nie mogę znaleźć rozwiązania mojego problemu, utknąłem w tym przez wiele godzin.

jestem usings Oracle AQs:

 Dbms_Aqadm.Create_Queue_Table(Queue_Table  => 'ITEM_EVENT_QT', 
            Queue_Payload_Type => 'ITEM_EVENT', 
            Multiple_Consumers => TRUE); 

     Dbms_Aqadm.Create_Queue(Queue_Name   => 'ITEM_EVENT_QUEUE', 
           Queue_Table   => 'ITEM_EVENT_QT', 
           Max_Retries   => 5, 
           Retry_Delay   => 0, 
           Retention_Time  => 432000, -- 5 DAYS 
           Dependency_Tracking => FALSE, 
           COMMENT    => 'Item Event Queue'); 
     -- START THE QUEUE 
     Dbms_Aqadm.Start_Queue('ITEM_EVENT_QUEUE'); 
     -- GRANT QUEUE PRIVILEGES 
     Dbms_Aqadm.Grant_Queue_Privilege(Privilege => 'ALL', 
             Queue_Name => 'ITEM_EVENT_QUEUE', 
             Grantee  => 'PUBLIC', 
             Grant_Option => FALSE); 
    END; 

Oto jeden z moich subskrybentów:

Dbms_Aqadm.Add_Subscriber(Queue_Name => 'ITEM_EVENT_QUEUE', 
          Subscriber => Sys.Aq$_Agent('ITEM_SUBSCRIBER_1', 
                 NULL, 
                 NULL), 
          rule => 'tab.user_data.header.thread_no = 1'); 

    Dbms_Aq.Register(Sys.Aq$_Reg_Info_List(Sys.Aq$_Reg_Info('ITEM_EVENT_QUEUE:ITEM_SUBSCRIBER_1', 
                  Dbms_Aq.Namespace_Aq, 
                  'plsql://ITEM_API.GET_QUEUE_FROM_QUEUE', 
                  HEXTORAW('FF'))),1); 

Rejestracja abonenta:

Gdy pewnego zdarzenia w moim DB, I” m przy użyciu wyzwalacza, aby dodać "zdarzenie" do mojego AQ, wywołując poniższą procedurę z mojego pakietu ITEM_API:

PROCEDURE ADD_EVENT_TO_QUEUE(I_EVENT  IN ITEM_EVENT, 
           O_STATUS_CODE OUT VARCHAR2, 
           O_ERROR_MSG OUT VARCHAR2) IS 

    ENQUEUE_OPTIONS DBMS_AQ.ENQUEUE_OPTIONS_T; 
    MESSAGE_PROPERTIES DBMS_AQ.MESSAGE_PROPERTIES_T; 
    MESSAGE_HANDLE  RAW(16); 
    EVENT    ITEM_EVENT; 
    HEADER_PROP  HEADER_PROPERTIES; 
    BEGIN 
    EVENT        := I_EVENT; 
    EVENT.SEQ_NO      := ITEM_EVENT_SEQ.NEXTVAL; 
    ENQUEUE_OPTIONS.VISIBILITY   := DBMS_AQ.ON_COMMIT; 
    ENQUEUE_OPTIONS.SEQUENCE_DEVIATION := NULL; 
    MESSAGE_PROPERTIES.PRIORITY  := 1; 
    MESSAGE_PROPERTIES.DELAY   := DBMS_AQ.NO_DELAY; 
    MESSAGE_PROPERTIES.EXPIRATION  := DBMS_AQ.NEVER; 
    HEADER_PROP      := HEADER_PROPERTIES(1); 
    EVENT.HEADER      := HEADER_PROP; 
    DBMS_AQ.ENQUEUE(QUEUE_NAME   => 'ITEM_EVENT_QUEUE', 
        ENQUEUE_OPTIONS => ENQUEUE_OPTIONS, 
        MESSAGE_PROPERTIES => MESSAGE_PROPERTIES, 
        PAYLOAD   => EVENT, 
        MSGID    => MESSAGE_HANDLE); 
    EXCEPTION 
    WHEN OTHERS THEN 
     ERROR_HANDLER.LOG_ERROR(NULL, 
           EVENT.ITEM, 
           EVENT.SEQ_NO, 
           SQLCODE, 
           SQLERRM, 
           O_STATUS_CODE, 
           O_ERROR_MSG); 
     RAISE; 
    END ADD_EVENT_TO_QUEUE; 

Działa to, ponieważ kiedy sprawdzam tabelę AQ, mogę znaleźć "zdarzenie", jednak moja metoda usuwania nie jest zapisywana, jak widać na poniższym obrazie, nie ma żadnej DEQ_TIME.

Oto moja metoda rozkolejkowania, również z moim pakiecie ITEM_API:

PROCEDURE GET_QUEUE_FROM_QUEUE(CONTEXT RAW, 
           REGINFO SYS.AQ$_REG_INFO, 
           DESCR SYS.AQ$_DESCRIPTOR, 
           PAYLOAD RAW, 
           PAYLOADL NUMBER) IS 

    R_DEQUEUE_OPTIONS DBMS_AQ.DEQUEUE_OPTIONS_T; 
    R_MESSAGE_PROPERTIES DBMS_AQ.MESSAGE_PROPERTIES_T; 
    V_MESSAGE_HANDLE  RAW(16); 
    I_PAYLOAD   ITEM_EVENT; 
    L_PROC_EVENT   BOOLEAN; 
    O_TARGETS   CFG_EVENT_STAGE_TBL; 
    O_ERROR_MSG   VARCHAR2(300); 
    O_STATUS_CODE  VARCHAR2(100); 
    BEGIN 
    R_DEQUEUE_OPTIONS.MSGID   := DESCR.MSG_ID; 
    R_DEQUEUE_OPTIONS.CONSUMER_NAME := DESCR.CONSUMER_NAME; 
    R_DEQUEUE_OPTIONS.DEQUEUE_MODE := DBMS_AQ.REMOVE; 
    --R_DEQUEUE_OPTIONS.WAIT   := DBMS_AQ.NO_WAIT; 
    DBMS_AQ.DEQUEUE(QUEUE_NAME   => DESCR.QUEUE_NAME, 
        DEQUEUE_OPTIONS => R_DEQUEUE_OPTIONS, 
        MESSAGE_PROPERTIES => R_MESSAGE_PROPERTIES, 
        PAYLOAD   => I_PAYLOAD, 
        MSGID    => V_MESSAGE_HANDLE); 
    IF I_PAYLOAD IS NOT NULL THEN 
     L_PROC_EVENT := PROCESS_EVENT(I_PAYLOAD, 
            O_TARGETS, 
            O_STATUS_CODE, 
            O_ERROR_MSG); 
    END IF; 
    EXCEPTION 
    WHEN OTHERS THEN 
     ERROR_HANDLER.LOG_ERROR(NULL, 
           NULL, 
           NULL, 
           SQLCODE, 
           SQLERRM, 
           O_STATUS_CODE, 
           O_ERROR_MSG); 
     RAISE; 
    END GET_QUEUE_FROM_QUEUE; 

robię coś źle? Jak mogę to naprawić? Myślę, że może być problem z moją rejestracją subskrybenta, ale nie jestem pewien.

EDYCJA: Właśnie odkryłem, że jeśli usunę subskrybentów i rejestr, a następnie ponownie je dodam, zostaną one usunięte z kolejki wszystkich wiadomości. Howerver, jeśli kolejne wydarzenie zostanie zmontowane, pozostanie tam w sposób nieokreślony (lub dopóki nie usuniesz i nie doda abonentów ponownie):

Rekord o stanie 0 i nr DEQ_TIME jest nowy.

Czy potrzebuję terminarza lub coś w tym stylu?

EDIT: Dodałem propagację planującego do mojego AQ:

DBMS_AQADM.SCHEDULE_PROPAGATION('ITEM_EVENT_QUEUE'); 

a nawet dodaje pole next_time:

DBMS_AQADM.SCHEDULE_PROPAGATION('ITEM_EVENT_QUEUE', SYSDATE + 30/86400); 

nadal nie działa. Jakieś sugestie? Zgaduję, że powiadomienia AQ nie działają, a moja procedura zwrotna nigdy nie jest wywoływana. Jak mogę to naprawić?

EDIT: ja zdjąłem procedurę z pakietu tylko w celach testowych, więc moi koledzy z zespołu mogą skompilować pakiet ITEM_API (nie wiem, czy rekompilacji pakietu, może lub nie może mieć wpływu na proces usuwania). Nadal nie działa.

Answer 1

Utwórz blok kodu i uruchom następujące:

DECLARE 
    dequeue_options  DBMS_AQ.dequeue_options_t; 
    message_properties DBMS_AQ.message_properties_t; 
    message_handle  RAW (16); 
    I_PAYLOAD   ITEM_EVENT; 
    no_messages exception; 
    msg_content   VARCHAR2 (4000); 
    PRAGMA EXCEPTION_INIT (no_messages, -25228); 
BEGIN 
    dequeue_options.wait := DBMS_AQ.NO_WAIT; 
    dequeue_options.consumer_name := 'ITEM_SUBSCRIBER_1'; 
    dequeue_options.navigation := DBMS_AQ.FIRST_MESSAGE; 
LOOP  
DBMS_AQ.DEQUEUE (queue_name   => 'ITEM_EVENT_QUEUE', 
         dequeue_options  => dequeue_options, 
         message_properties => message_properties, 
         payload    => I_PAYLOAD, 
         msgid    => message_handle 
        ); 
END LOOP; 
    EXCEPTION 
    WHEN no_messages 
    THEN 
    DBMS_OUTPUT.PUT_LINE ('No more messages left'); 
END; 

Daj mi znać, co się dzieje z skolejkowany wiadomości.

Powinieneś mieć stolik, w którym zapisujesz dane.

Można również spróbować dodać tabelę enqueud w agencie, a następnie określić agenta do tabeli usuwania.

DECLARE 
    aSubscriber sys.aq$_agent; 
BEGIN 
    aSubscriber := sys.aq$_agent('ITEM_SUBSCRIBER_1', 
          'ITEM_EVENT_QUEUE', 
          0); 
    dbms_aqadm.add_subscriber 
(queue_name  => 'ITEM_EVENT_QUEUE' 
    ,subscriber  => aSubscriber); 
END; 
/

Answer 2

I wobec tego samego problemu, ale to rozwiązano po zmianie tych parametrów 2 dB:

1. job_queue_processes (musi być> niż 0)
2. aq_tm_processes (AutoTuning)

Mam nadzieję, że to pomaga.