Enterprise Service Bus na Was 6

Termin Enterprise Service Bus (ESB) oznacza szynę dostępową dla różnego rodzaju usług. Bardziej szczegółowo termin ten postaram się opisać w części związanej z Message Broker, który jest, wydaje się, bardziej naturalnym narzędziem dla ESB. W WebSphere Application Server od wersja 6 została zaimplementowana szyna danych o nazwie Service Integration Bus (SIB).

Podstawowymi elementami SI w WAS są:

Bus
Bus to logiczny element określający szynę danych. W celi Was lub na pojedynczym serwerze może być wiele połączonych pomiędzy sobą SIB. Każdy SIB zawiera członków i mogą to być albo serwery aplikacji albo klastry WAS.

Message Engine
Message Engine (ME) to element realizujący w Was analogiczna funkcje jaką pełni menadżer kolejek MQ. Zajmuje się on zarządzaniem zasobów szyny danych. Message Engine istnieje jeden dla serwera aplikacyjnego i jeden lub więcej dla klastra WAS. Obiekt ten jest tworzony automatycznie przy dodawaniu członka do szyny danych i tak samo usuwany przy usuwaniu członka z szyny.

Destinatioon
Destinatioon wskazuje na fizyczny obiekt Queue lub Topic zarządzany wewnątrz jednego lub wielu ( w przypadku klastra ) Message Engine. Dodatkowo może on być typem Alias lub Foreign (wskazuje na destination w innej szynie).

Wszystkie komunikaty, które zarządzane dochodzą do ME oraz subskrypcje na topic przechowywane są w bazie danych. Standardowo jest to baza Cloudscape, którą można zastąpić przez inne silniki danych.

Inną ciekawą funkcjonalnością wprowadzoną w SIB jest Mediation. Mediation za pomocą odpowiedniego bean'a pozwala na manipulacje treścią komunikatu, zmianę jego docelowej kolejki. Obiekt ten "podpina" się w panelu administarcyjnym pod konkretny Destination.

Współpraca z MQ

W związku z tym, że SIB jest zamknięty tylko dla aplikacji J2EE wskazane jest aby otworzyć go dla aplikacji nie J2EE. Taką możliwość daje nam użycie MQ Link w SIB. W przypadku MQ link ME zachowuje się jak menadżer kolejek i tak przez standardowy menadżer MQ jest postrzegany. Dla poprawnej komunikacji pomiędzy SIB a MQ należy dodatkowo utworzyć odpowiednie kanały sender i reciever po obu stronach oraz kolejkę transmisyjną po stronie MQ.

Przykładowy scenariusz komunikacji
W tej części postaram się opisać pełną konfiguracje na przykładzie prostej aplikacji. Scenariusz polega na tym , że komunikat umieszczony w kolejce MQ zostanie przesłany do SIB. Wewnątrz WAS komunikat ten będzie odczytany przez MDB, który wyświetli treść komunikatu i wyśle odpowiedź, która w ostateczności pojawi się jako komunikat w kolejce MQ określonej w polach ReplyToQueue i ReplyToQmgr pierwszego wysłanego komunikat.

Szyna danych
Pierwszym elementem jest stworzenie szyny danych. W panelu administracyjnym w zakładce Service integration -> Buses tworzymy szynę o nazwie MKBus oraz dodajemy do niego członków w postaci serwera aplikacyjnego lub klastra WAS.

Foreign bus
W związku z tym, że będziemy się komunikowali z MQ, musimy stworzyć zewnętrzna szynę danych typu MQ. Aby to zrobić na poziomie szyny MKBus wybieramy Foreign buses i tworzymy zewnetrzną szyne typu MQ o nazwie QM1 .

MQ link
Połączenie z MQ definiuje się poprzez zakładkę Buses -> Nazwa szyny -> Messaging Engine -> Nazwa ME -> WebSphere MQ links. Tu ustawiamy nazwę na np. QM1 ,Foreign bus na QM1 , Queue manager name na MKBus . Ten ostatni parametr określa nazwę jaka ME będzie się przedstawiał przy komunikacji z MQ i przez którą będzie identyfikowany.

Dalej wewnątrz MK link tworzymy kanały sender i reciever. Dla kanału sender ustawiamy parametry Sender MQ channel name na MKBus/QM1 , Host name na localhost, port na 1414. Dla kanału receiver ustawiamy parametry Receiver MQ channel name na QM1/MKBus

Oznacza to, że będziemy się komunikować z menadżerem QM1 zainstalowanym lokalnie na porcie 1414. Z drugiej strony na menadżerze MQ musimy utworzyć odpowiednie kanały oraz kolejkę transmisyjną w celu komunikacji z ME (emulującym menadżer kolejek MQ o nazwie MKBus)

Destinations
W szynie danych mamy zakładkę Destinations. W niej tworzymy kolejkę QIN określającą kolejkę w ESB, z której MDB będzie czytał dane oraz alias kolejki o nazwie QueueonMQ , który wskazuje na docelową kolejkę MQ. W przypadku pierwszej kolejki ustawiamy tylko jej nazwę na QIN oraz typ na Queue. W przypadku aliasu ustawiamy identyfikator na QueueonMQ, wybieramy Bus na MKBus wybieramy Target bus na QM1 oraz ustawiamy Target identifier na QOUT3@QM1 .

Obiekty JMS
W tym miejscu założymy obiekty JMS użyte w przykładzie. Wszystkie obiekty zakładamy na standardowym providerze znajdującym się w zakładce Resources -> JMS Providers -> Default messaging. Pierwszym obiektem, jaki tu założymy będzie JMS connection factory używany przez MDB do wysyłki odpowiedzi. ustawimy mu nazwę na qcfbus, JNDI name na jms/qcfbus , bus name na MKBus .

Następnie zakładamy dwa obiekty JMS queue. W pierwszym o nazwie Mqqueue ustawiamy parametry JNDI name na jms/mqqueue , Bus name wybieramy MKBus , Queue name wybieramy QueueonMQ W drugim o nazwie MyMDBQueue ustawiamy parametry JNDI name na jms/MyMDBQueue , Bus name wybieramy MKBus , Queue name wybieramy QIN

W tym miejscu musimy pamiętać o jeszcze jednym obiekcie JMS activation specification. Ustawiamy jego nazwę na MyMDBQueue_Act_Spec , JNDI name na eis/MyMDBQueue_Act_Spec w polu Destinaion type wybieramy Queue, w polu Destinaion JNDI name ustawiamy jms/MyMDBQueue w polu Bus name wybieramy MKBus . Obiekt ten odpowiadam za połączenie MDB z konkretną kolejką, z której MDB ma odczytywać komunikaty.

Ostatnim elementem będzie ustawienie JMS activation specification dla aplikacji. Aby to zrobić przechodzimy do zakładki Enterprise Applications -> nazwa aplikacji -> Provide listener bindings for message driven bean i tam w polu Activation Specification -> JNDI name wpisujemy eis/MyMDBQueue_Act_Spec

Dla przykładu poniżej fragment kodu MDB:

public void onMessage(javax.jms.Message msg) {
                
                Destination dest= null;
                Destination reply= null;
                System.out.println("przyszedl komunikat \n"+msg);
                try {
                        dest=msg.getJMSDestination();
                        System.out.println("dest "+dest);
                        reply=msg.getJMSReplyTo();
                        System.out.println("reply "+reply);
                        sendResponse(msg);
                } catch (Exception e) {
                        System.out.print(e);
                        e.printStackTrace();
                }
                
        }
        
        public void sendResponse(javax.jms.Message msg) throws Exception
        {
                Destination des = msg.getJMSReplyTo();
        
        Context     ctx       =new InitialDirContext( );

                 
        ConnectionFactory factory = (ConnectionFactory)ctx.lookup("jms/qcfbus");
        
        Connection connection = factory.createConnection();
        connection.start();
                
        boolean transacted = false;
        Session  session = connection.createSession( transacted,
                        Session.AUTO_ACKNOWLEDGE);
        

        Queue desQueue = (Queue)ctx.lookup( "jms/mqqueue" );
        
        
        MessageProducer producer = session.createProducer(desQueue);
        TextMessage msg1 = session.createTextMessage();
        TextMessage textmessage= (TextMessage)msg;
        
        msg1.setText("odp:"+textmessage.getText());
        msg1.setJMSCorrelationID(msg.getJMSMessageID()); /* set correlation id of */
        msg1.setJMSReplyTo(des);                                 /* reply to request's    */
            /* message id            */
        producer.send(msg1);
        session.close();
        connection.close();
        
        }

Powrót