puetz_SOA_AQ_XML

TAG DER DATENBANKEN
20.06.2013
Udo Pütz
Dr. Andreas Kumpf
Advanced Queues und XML-DB
Integration von Oracle in einer SOA
mit Hilfe von asynchroner Kommunikation und
XML-Verarbeitung
Übersicht
Motivation
ESB
XML-DB
Advanced Queues
Gewachsene IT-Landschaft
• Vorhandene Legacy-Systeme
• Verschiedene Technologien
• Viele Punkt-zu-Punkt Schnittstellen
(intern, extern)
• Oft: Datenbanken-Sst., Datei-Sst., propr. APIs,
etc.
• synchrone und asynchrone Kommunikation
• Einzeloperationen und Batchverarbeitung
Service Orientierte Architektur (SOA)
• Einführung eines zentralen Enterprise Service
Bus (ESB)
• Verwaltung von Endpunkten
(Service-Virtualisierung)
• Eine Schnittstelle pro System zwischen ESB
und Legacy System
• ESB routet Messages zwischen den Systemen
• Standard: XML-Austausch über WebServices
(SOAP) und Queues (z.B. JMS)
Oracle Schnittstellen
• Reine DML Schnittstelle (select, insert,
update, delete)
• OCI, OCCI, ODBC, JDBC
• Aufruf von Stored Procedures (synchron)
• Verwendung von WebServices (synchron)
• Queueing von Messages (asnychron)
• Verarbeitung von XML
Business Interface
Option 1
XMLs
ESB
Legacy System
BI
DB
„reincompiliert“
Option 2
XMLs
ESB
Option 3
API
BI
DB
DB-Sst.
XMLs
ESB
Legacy System
BI
DB
Legacy System
SOA-Beispielszenario
ESB
enqueue
(AQ) ESB_2_BI
dequeue
XML
ESB
dequeue
(AQ) BI_2_ESB
enqueue
Business
Interface
(PL/SQL)
Legacy
System
XML-DB
XML
DB
XML-Verarbeitung
• XML wird binär gespeichert (innerhalb BLOB)
• XML wird geparsed und relational gespeichert
• XML wird als XMLType-Spalte abgelegt
(d.h. als DOM)
Oracle XML-DB
•
•
•
•
•
•
Seit 8i XML-Funktionen und Schnittstellen in der DB
Speichern und Manipulieren von XML
Datentyp XMLType (seit 9iR1)
XPath
XML DB Repository für XML Schemata (seit 9iR2)
Weiterer Ausbau der Funktionalität in 10g (Performance,
direct load...)
• XMLElement, XMLAgg, XMLForest,XMLConcat,XMLAgg
erzeugen XML aus einem relationalem Modell
• Extract, ExistsNode, Extractvalue zum Selektieren aus
XMLType
Beispiel: XML-Erzeugung
FILIALE
FilNo
Name
Ort
1
Kundencenter
Koeln
2
Verwaltung
Bonn
MITARBEITER
PersonalNo
Name
FilNo
10
Fox Mulder
1
11
Dana Scully
1
12
Olivia Dunham
2
13
Walter Bishop
2
SELECT f.name filial_name,
f.ort einsatz_ort,
m.name mitarbeiter_name
FROM
filiale f, mitarbeiter m
WHERE m.filNo = f.filNo
filial_name
einsatz_ort
mitarbeiter_name
Kundencenter
Koeln
Fox Mulder
Kundencenter
Koeln
Dana Scully
Verwaltung
Bonn
Olivia Dunham
Verwaltung
Bonn
Walter Bishop
Beispiel: XML-Erzeugung
SELECT
XMLElement("FILIALE", XMLAttributes(f.name AS "name"),
XMLElement("ANGESTELLTE",
XMLForest(f.ort AS "EINSATZORT"),
(SELECT XMLAgg(XMLElement("ANGESTELLTER",
XMLAttributes(m.name AS "name")))
FROM mitarbeiter m
WHERE m.filNo = f.filNo
)
<FILIALE name="Kundencenter">
)
<ANGESTELLTE>
) report_xml
<EINSATZORT>Koeln</EINSATZORT>
FROM
filiale f ► 2 Zeilen als Ergebnis
<ANGESTELLTER name="Fox Mulder"/>
<ANGESTELLTER name="Dana Scully"/>
</ANGESTELLTE>
</FILIALE>
<FILIALE name="Verwaltung">
<ANGESTELLTE>
<EINSATZORT>Bonn</EINSATZORT>
<ANGESTELLTER name="Olivia Dunham"/>
<ANGESTELLTER name="Walter Bishop"/>
</ANGESTELLTE>
</FILIALE>
Beispiel: XML-Selektion über XPath
SELECT COUNT(*)
FROM
dual
WHERE existsNode(XMLType('
<FILIALE name="Kundencenter">
<ANGESTELLTE>
<EINSATZORT>Koeln</EINSATZORT>
<ANGESTELLTER name="Fox Mulder"/>
<ANGESTELLTER name="Dana Scully"/>
</ANGESTELLTE>
</FILIALE>'),
'//FILIALE') = 1
SELECT COUNT(*)
FROM
dual
WHERE extractValue(XMLType('
<FILIALE name="Kundencenter">
<ANGESTELLTE>
<EINSATZORT>Koeln</EINSATZORT>
<ANGESTELLTER name="Fox Mulder"/>
<ANGESTELLTER name="Dana Scully"/>
</ANGESTELLTE>
</FILIALE>'),
'//FILIALE/ANGESTELLTE/EINSATZORT/text()') = 'Koeln'
Oracle XML Query Rewrite
• Optimierung beim Selektieren und
Extrahieren von Daten
• Ohne Query Rewrite wird das XML aus der DB
gelesen, dann erst in DOM gewandelt und
weiter extrahiert/manipuliert
• Mit Query Rewrite wird indizierter Zugriff auf
das relationale DB-Modell möglich
Bsp. Query Rewrite
CREATE VIEW mitarbeiter_liste AS
SELECT
XMLElement("FILIALE", XMLAttributes(f.name AS "name"),
XMLElement("ANGESTELLTE",
XMLForest(f.ort AS "EINSATZORT"),
(SELECT XMLAgg(XMLElement("ANGESTELLTER",
XMLAttributes(m.name AS "name")))
FROM mitarbeiter m
WHERE m.filNo = f.filNo
)
)
) report_xml
FROM filiale f
<FILIALE name="Kundencenter">
<ANGESTELLTE>
<EINSATZORT>Koeln</EINSATZORT>
<ANGESTELLTER name="Fox Mulder"/>
<ANGESTELLTER name="Dana Scully"/>
</ANGESTELLTE>
</FILIALE>
<FILIALE name="Verwaltung">
<ANGESTELLTE>
<EINSATZORT>Bonn</EINSATZORT>
<ANGESTELLTER name="Olivia Dunham"/>
<ANGESTELLTER name="Walter Bishop"/>
</ANGESTELLTE>
</FILIALE>
SELECT *
FROM
mitarbeiter_liste
WHERE existsNode(report_xml,
'//FILIALE/ANGESTELLTE/ANGESTELLTER[@name="Walter Bishop"]') = 1
SELECT *
FROM
mitarbeiter_liste WHERE
mitarbeiter.name = ‘Walter Bishop‘
Oracle Advanced Queues (AQ)
•
•
•
•
FIFO: first in first out
Sender macht enqeue
Empfänger macht dequeue
Oracle stellt sicher, dass Messages nicht
verlorengehen durch:
– Persistenz
– Transaktionshandling
Advanced Queue
ORACLE AQ
enqueue
Diverse Schnittstellen:
• PL/SQL
• VB (Oracle Objects for OLE)
• Java (oracle.AQ, oracle.jms)
• HTTP, HTTPS, SMTP
…
dequeue
Features wie:
• Exception handling
• Rules
• Transformation
• Statistic Views
AQ: Point to Point
ORACLE AQ
Sender
enqueue
dequeue
Consumer
Publish-Subscribe
Sender 1
Sender n
pu
bli
sh
lish
b
pu
ORACLE AQ
be
i
r
sc
b
su
sub
sc
ribe
Consumer 1
Consumer n
Package DBMS_AQADM
• Administrieren von AQs
• Anlegen / Entfernen von Queues und Queue
Tables
• Starten / Stoppen von Queue
• Festlegen der Prioritätssteuerung
AQ: Codebeispiel
-- Definition des Payload-Typs
CREATE OR REPLACE
TYPE typ_my_aq_message AS OBJECT(message XMLTYPE)
-- Erstellen der Queue-Table
EXECUTE DBMS_AQADM.CREATE_QUEUE_TABLE(
QUEUE_TABLE => 'QT_MY_QUEUE_1',
QUEUE_PAYLOAD_TYPE=> 'typ_my_aq_message');
-- Erstellen der Queue
EXECUTE DBMS_AQADM.CREATE_QUEUE (
QUEUE_NAME=>'QU_MY_QUEUE_1',
QUEUE_TABLE=>'QT_MY_QUEUE_1');
-- Starten/Stoppen der Queue
EXECUTE DBMS_AQADM.START_QUEUE(
QUEUE_NAME=>'QU_MY_QUEUE_1');
Package DBMS_AQ
• PL/SQL Schnittstelle zu AQ
• Enqueue und dequeue von Messages
• Exception Handling
Codebeispiel Enqueue
DECLARE
v_myXmlString
xml_message
VARCHAR2(1000);
XMLTYPE;
mymessage
TYP_MY_AQ_MESSAGE;
message_properties
dbms_aq.message_properties_t;
enqueue_options
dbms_aq.enqueue_options_t;
msgid
RAW(16);
BEGIN
v_myXmlString := '
<MyMessage>
<MyText>This is my Data</MyText>
</MyMessage>';
xml_message := xmltype(v_myXmlString);
mymessage := TYP_MY_AQ_MESSAGE(NULL);
mymessage.message := xml_message.getClobVal();
dbms_aq.enqueue(
queue_name
message_properties
enqueue_options
payload
msgid
END;
=>
=>
=>
=>
=>
'QU_MY_QUEUE_1',
message_properties,
enqueue_options,
mymessage,
msgid);
Codebeispiel Dequeue
DECLARE
v_myXmlString
message_xml
VARCHAR2(1000);
XMLTYPE;
mymessage
TYP_MY_AQ_MESSAGE;
message_properties
dbms_aq.message_properties_t;
dequeue_options
dbms_aq.dequeue_options_t;
msgid
RAW(16);
BEGIN
mymessage := TYP_MY_AQ_MESSAGE(NULL);
msgid := NULL;
message_xml := NULL;
--vaqDequeueOpt.wait := DBMS_AQ.FOREVER); --wartet
dequeue_options.wait := (10); --gibt nach 10 sec auf
dbms_aq.dequeue(
queue_name
message_properties
dequeue_options
payload
msgid
=>
=>
=>
=>
=>
'QU_MY_QUEUE_1',
message_properties,
dequeue_options,
mymessage,
msgid);
message_xml := mymessage.message;
v_myXmlString := message_xml.getStringVal();
dbms_output.put_line (substr(v_myXmlString, 1, 256));
END;
Dequeue Exception Handling
DECLARE
MyDequeueNotEnabled EXCEPTION;
MyDequeueTimout
EXCEPTION;
PRAGMA EXCEPTION_INIT(MyDequeueNotEnabled, -25226); -- Code ist
festgelegt
PRAGMA EXCEPTION_INIT(MyDequeueTimout, -25228);
-- Code ist
festgelegt
...
BEGIN
...
dequeue...
...
EXCEPTION
WHEN MyDequeueTimout THEN
dbms_output.put_line('Zu lange auf Queue gewartet');
WHEN MyDequeueNotEnabled THEN
dbms_output.put_line('Queue ist nicht gestartet');
END;
AQ: Publish Subscribe Beispiel Anlegen
-- Erstellen der Queue-Table
EXECUTE DBMS_AQADM.CREATE_QUEUE_TABLE(
QUEUE_TABLE => 'QT_MY_TOPIC_1',
QUEUE_PAYLOAD_TYPE=> ‘RAW‚
multiple_consumers => TRUE);
-- Erstellen der Queue
EXECUTE DBMS_AQADM.CREATE_QUEUE (
QUEUE_NAME=>'QU_MY_TOPIC_1',
QUEUE_TABLE=>'QT_MY_TOPIC_1');
-- Starten/Stoppen der Queue
EXECUTE DBMS_AQADM.START_QUEUE(
QUEUE_NAME=>'QU_MY_TOPIC_1');
AQ: Publish Subscribe Beispiel Subscriber
-- Definition der Subscriber
DECLARE
subscriber sys.aq$_agent;
BEGIN
subscriber := sys.aq$_agent(
'SUBSCRIBER1',
'QU_MY_TOPIC_1',
null);
dbms_aqadm.add_subscriber(
queue_name
=> 'QU_MY_TOPIC_1',
subscriber
=> subscriber,
delivery_mode => dbms_aqadm.buffered);
END;
DECLARE
subscriber sys.aq$_agent;
BEGIN
subscriber := sys.aq$_agent(
'SUBSCRIBER2',
'QU_MY_TOPIC_1',
null);
dbms_aqadm.add_subscriber(
queue_name
=> 'QU_MY_TOPIC_1',
subscriber
=> subscriber,
delivery_mode => dbms_aqadm.buffered);
END;
AQ: Publish Subscribe Beispiel Enqueue
CREATE OR REPLACE PROCEDURE do_enqueue(p_hexamsg varchar2) 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', 'QU_MY_TOPIC_1', NULL);
recipients(2) := sys.aq$_agent('SUBSCRIBER2', 'QU_MY_TOPIC_1', NULL);
message_properties.recipient_list := recipients;
enqueue_options.delivery_mode := dbms_aq.PERSISTENT;
dbms_aq.enqueue(
queue_name
enqueue_options
message_properties
payload
msgid
COMMIT;
END;
=>
=>
=>
=>
=>
'QU_MY_TOPIC_1',
enqueue_options,
message_properties,
message,
message_handle);
AQ: Publish Subscribe Beispiel Dequeue
CREATE OR REPLACE PROCEDURE do_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);
MyDequeueTimout
EXCEPTION;
PRAGMA EXCEPTION_INIT(MyDequeueTimout, -25228);
BEGIN
dequeue_options.wait
:= dbms_aq.no_wait;
dequeue_options.consumer_name := p_subscriber;
dequeue_options.navigation
:= dbms_aq.first_message;
dequeue_options.delivery_mode := dbms_aq.PERSISTENT;
dbms_aq.dequeue(
queue_name
=> 'QU_MY_TOPIC_1',
dequeue_options
=> dequeue_options,
message_properties => message_properties,
payload
=> message,
msgid
=> message_handle);
dbms_output.put_line('Message: '|| hextoraw(message) );
EXCEPTION WHEN MyDequeueTimout THEN
dbms_output.put_line('Queue empty');
COMMIT;
END;
AQ: Publish Subscribe Beispiel Verwendung
exec do_enqueue('1');
exec do_enqueue('2');
exec do_enqueue('3');
exec
exec
exec
exec
exec
exec
exec
exec
do_dequeue('SUBSCRIBER1');
do_dequeue('SUBSCRIBER1');
do_dequeue('SUBSCRIBER1');
do_dequeue('SUBSCRIBER1');
do_dequeue('SUBSCRIBER2');
do_dequeue('SUBSCRIBER2');
do_dequeue('SUBSCRIBER2');
do_dequeue('SUBSCRIBER2');
->
->
->
->
->
->
->
->
“1“
“2“
“3“
“Queue empty“
“1“
“2“
“3“
“Queue empty“
Kontakt
telexiom AG
Spichernstr. 6 B
50672 Köln
www.telexiom.de
www.health-care24.de
facebook.com/telexiom
Tel. 0221 888840-0
Fax. 0221 888840-40
[email protected]
xing.com
linkedin.com