Oracle英文版培训课件之Implement Streams：les18

Message QueuingConceptsObjectivesAftercompletingthislesson,youshouldbeabletodothefollowing:ConfiguresecurequeueusersforenqueuinganddequeuingmessagesConfigureamessagingclientfordequeuingeventsDescribetheprocessofenqueuinganddequeuingeventsConfiguremessagenotificationManageAdvancedQueuingqueues,queuetables,propagationschedules,andtransformationsbyusingEnterpriseManagerEnqueuingEventsEventscanbeenqueuedintoastagingqueueinthreeways:AcaptureprocessenqueuesDMLandDDLchangesasLCRevents.AuserapplicationenqueuesusermessagesorLCRsaseventsoftypeSYS.AnyData.

Anenqueuedirectiveinstructsanapplyprocesstoenqueuecertaineventsintothestagingqueue.SecureQueueUsersForausertobeabletoenqueueordequeuemessagesfromasecureagentqueue,youmustconfiguretheuserasasecurequeueuser.UseDBMS_STREAMS_ADM.SET_UP_QUEUEtoconfigureasecurequeueuser.BEGINDBMS_STREAMS_ADM.SET_UP_QUEUE(queue_table=>'ix.streams_queue_table',queue_name=>'streams_queue',queue_user=>'ix');END;/SubscriptionsandRecipientListsTheSYS.AnyDataqueueisamulticonsumerqueuethatallowsconsumptionofamessagebymorethanoneconsumer.Alistofconsumersforamessageisidentifiedbyusing:SubscriptionAddasubscriptiontoaqueuebyusingtheDBMS_AQADM.ADD_SUBSCRIBERprocedureRecipientlistSpecifiedintherecipient_listmessagepropertywhenthemessageisenqueuedProceduresforConfiguringAQAgentsDBMS_AQADM.CREATE_AQ_AGENT(agent_nameINVARCHAR2,enable_httpINBOOLEANDEFAULTFALSE,enable_smtpINBOOLEANDEFAULTFALSE,enable_anypINBOOLEANDEFAULTFALSE);DBMS_AQADM.ENABLE_DB_ACCESS(agent_nameINVARCHAR2,db_usernameINVARCHAR2)DBMS_AQADM.ADD_SUBSCRIBER(queue_nameINVARCHAR2,subscriberINsys.aq$_agent,ruleINVARCHAR2DEFAULTNULL,transformationINVARCHAR2DEFAULTNULL);CreatingaSubscriberToconfigureasecurequeueuserasasubscriber,thusenablingthemtodequeuemessagesoreventsfromaqueue,usethe:SYS.AQ$_AGENTtypetospecifytheAQagentDBMS_AQADM.ADD_SUBSCRIBERproceduretospecifythattheagentisasubscribertothequeueDECLAREsubscriberSYS.AQ$_AGENT;BEGINsubscriber:=SYS.AQ$_AGENT('IX',NULL,NULL);SYS.DBMS_AQADM.ADD_SUBSCRIBER(queue_name=>'ix.streams_queue',subscriber=>subscriber);END;/MessagingClientUserAmessagingclient:Consumesuser-enqueuedeventsfromaSYS.AnyDataqueuebasedonrulesIsinvokedbyauserorapplicationIsassociatedwithonlyonedatabaseuserAmessagingclientuser:CreatesthemessagingclientusingproceduresintheDBMS_STREAMS_ADMpackageIsgrantedtheprivilegestodequeuefromthequeueCanbeassociatedwithmanymessagingclientsCreatingaMessagingClientDBMS_STREAMS_ADM.ADD_TABLE_RULES(table_name=>'oe.orders',streams_type=>'dequeue',streams_name=>'ORDERS_DEQ',queue_name=>'ix.streams_queue',source_database=>'',inclusion_rule=>true);Applicationdequeue

usingmessagingclientCreatingMessagingClientRulesDBMS_STREAMS_ADM.ADD_MESSAGE_RULE(message_type=>'ix.order_event_typ',rule_condition=>':msg.delivery_date<SYSDATE',streams_type=>'dequeue',streams_name=>'ORDERS_DEQ',queue_name=>'ix.streams_queue',inclusion_rule=>true);Applicationdequeue

usingmessagingclientUser-CreatedEventsMustbeexplicitlyenqueuedCanbeamessageofanydatatypeoruser-definedtypeMustbewrappedinaSYS.AnyDatatypetobeenqueuedinaSYS.AnyDataqueueCanalsobeaJavaMessageService(JMS)typemessageUser-EnqueuedEventsThemessagepayloadisoftypeSYS.AnyData.YoucanenqueuemessageswithdifferenttypesofpayloadsintoaStreamsstagingqueuebywrappingthemessageintheSYS.AnyDatatype:User-createdLCRsStandarddatatypesUser-definedobjecttypesSYS.AnyData

typeApplicationWrappingMessagePayloadsUsetheCONVERT<data_type>staticfunctionsoftheSYS.AnyDatatype,wheredata_typeisthetypeofobjecttowrap.Alldatatypesaresupportedexcept:CLOB,NCLOB,BLOB,orROWIDUser-definedtypeswithLOBattributes(includingCLOB,

NCLOB,andBLOB)NestedtableSYS.AnyData.ConvertObject(ix.order_event_typ(2424,3350,146,'Elia','Fawcett',4,TO_DATE('18-AUG-02')))StreamsMessageswithObjectTypesEachuser-definedtypethatisusedinamessagemustexistateverydatabasewherethemessagemaybestagedinaqueue.Thetypemusthavethesamename,owner,andformateachdatabase.Typeevolutionandinheritanceisnotsupported.Thetype’sobjectidentifier(OID)doesnothavetobethesameforeachdatabase.StreamsMessagesContainingLCRsAnexplicitlyenqueuedmessagecanbe:AnLCRwrappedintheSYS.AnyDatatypeAuser-definedtypethatcontainsanLCRasanattributeLCRwrappedin

SYS.AnyDatatypeidLCRObjecttypewithLCR

attributewrappedin

SYS.AnyDatatypeDequeuingMessagesinStreamsTheoperationofretrievingmessagesfromaqueueisknownasdequeuing.User-enqueuedeventsinthestagingqueuecanbedequeuedautomatically:Applyhandlesuser-enqueuedLCReventsinthesamemannerascapturedLCRevents.Applyconfiguredwithamessagehandlerdequeuesandprocessesnon-LCRevents.YoucanusetheDBMS_STREAMS_ADM.DEQUEUEorDBMS_AQ.DEQUEUEprocedurestoexplicitlydequeueuser-enqueuedevents.UnwrappingMessagePayloadsUsetheGET<datatype>staticfunctionsoftheSYS.AnyDatatype,where<datatype>isthetypeofobjecttounwrap.UsetheGETTYPEorGETTYPENAMEstaticfunctionsoftheSYS.AnyDatatypetodeterminethetypeofembeddeddata....IF(evt.GetType()=DBMS_TYPE.TYPECODE_VARCHAR2)

THENnum_var:=evt.GetVarchar2(string_var);ELSEIF(evt.GetTypeName()='SYS.LCR$_ROW_RECORD')THENnum_var:=evt.GetObject(rowLCR);...MessageHandlersforUserMessagesUsermessages(non-LCRevents)arealwaysprocessedbyamessagehandler.Themessagehandlermustperformanyneededdependencychecking.Therecanbeonlyasinglemessagehandlerperapplyprocess.Non-LCReventMessagehandlerMessageHandlersAmessagehandlerisauser-definedprocedurethatcanprocessnon-LCRusermessagesinacustomizedwayforyourenvironment.Youcanspecifyamessagehandlerforanapplyprocessbyusingthemessage_handlerparameterintheDBMS_APPLY_ADMpackage:CREATE_APPLYALTER_APPLYTheapplyprocesshandlesthedequeueoperation.Ifdependenciesexistbetweenthesemessagesinyourenvironment,setparallelismto1fortheapplyprocess.MessageHandler:ExampleCREATEORREPLACEPROCEDUREorder_events_mh

(evtSYS.AnyData)ISordorder_event_typ;dummyNUMBER;BEGINIF(evt.getTypeName='OE.ORDER_EVENT_TYP')THENdummy:=evt.getObject(ord);CASEWHENord.shipdays=0THEN<routebilling>;WHENord.shipdays<=3THEN<routepriority>;WHENord.shipdays>3THEN<routenormal>;ENDCASE;INSERTINTOoe.order_logVALUES(ord.order_number,ord.part_number,SYSTIMESTAMP);ENDIF;END;UsingaMessageHandlerCreateanapplyprocessthatisassociatedwiththedestinationqueue.Configurethemessagehandlerfortheapplyprocess.Createatleastonerulefortheapplyprocessrelatingtotheuser-enqueuedevent,andplaceitinthepositiveornegativeruleset.Starttheapplyprocess.ConfiguringMessageNotificationEXECUTE-DBMS_STREAMS_ADM.SET_MESSAGE_NOTIFICATION(-streams_name=>'ORDERS_DEQ',-notification_action=>'john.doe@',-notification_type=>'MAIL',-include_notification=>true,-queue_name=>'IX.STREAMS_QUEUE');MessagingclientNotificationSupportedNotificationTypesForURLnotifications,specifyaURLwithouttheprefixhttp://ForPL/SQLprocedurenotifications,specifyaPL/SQLprocedurename:Fore-mailnotifications,specifyane-mailaddress:notification_type=>'HTTP'notification_action=>':8080'notification_type=>'PROCEDURE'notification_action=>'oe.notify_addr_arrival'

notification_type=>'MAIL'notification_action=>'John.Doe@'

PurgingtheQueueTheDBMS_AQADM.PURGE_QUEUE_TABLEprocedure:RemovesmessagesfrompersistentqueuesIsusefulforbothsingle-consumerandmulticonsumerqueuesCanbecustomizedtopurgeonlymessagesthatmeetspecificconditionsPurgingtheQueueWhenpurgingeventsfromaqueue,youcanspecify:ApurgeconditionWhetheranexclusivelockisobtainedduringthepurgeoperationDECLAREpurge_optDBMS_AQADM.aq$_purge_options_t;BEGINpurge_opt.block:=TRUE;DBMS_AQADM.PURGE_QUEUE_TABLE(queue_table=>'STRMADMIN.STREAMS_QUEUE_TABLE',purge_condition=>'queue=''STREAMS_QUEUE''',purge_options=>purge_opt);END;PurgeConditions:ExamplesPurgealleventsinaqueuetable

Purgeallevent