W środowiskach rozproszonych moduły aplikacji mogą wspóldziałac poprzez wymianę komunikatów. Jest to dobry i ciekawy sposób programowania, pozwala bowiem uniezależniać kody modułów. Poznamy go na przykładzie JMS (Java Messaging Service).

1. Wprowadzenie

• klient-producent wysyła komunikat,
• klient-konsument odczytuje komunikat,
• obaj klienci nie muszą działać równocześnie,
• żaden z nich nie musi nic "wiedzieć" o budowie, kodzie itp. drugiego.

2. Architektura JMS

• serwis (JMS provider) implementuje interfejsy i dostarcza narzędzi administrowania,
• klienci - tworzą, wysyłają i odbierają komunikaty,
• obiekty administrowane - destynacje oraz fabryki połączeń (są zwykle tworzone i konfigurowane przez narzędzia administracyjne, ale również istnieje programistyczne API do tych celów)

• kolejka (Queue) -  Point-to-Point za pośrednictwem kolejki komunikatów, nadawca wysyła do określonej nazwanej kolejki (destynacji), odbiorca - odbiera komunikat z tej kolejki.

• temat (Topic) - zasada publikacji i subskrypcji.

3. Konfigurowanie obiektów administrowanych

  <Connectors>
    <Connector scheme="tcp">
      <ConnectionFactories>
        <ConnectionFactory name="ConnectionFactory" />
      </ConnectionFactories>
    </Connector>
    <Connector scheme="rmi">
      <ConnectionFactories>
        <QueueConnectionFactory name="JmsQueueConnectionFactory" />
        <TopicConnectionFactory name="JmsTopicConnectionFactory" />
      </ConnectionFactories>
    </Connector>
  </Connectors>

.....

  <AdministeredDestinations>
    <AdministeredTopic name="topic1">
      <Subscriber name="sub1" />
      <Subscriber name="sub2" />
    </AdministeredTopic>

    <AdministeredQueue name="queue1" />
    <AdministeredQueue name="queue2" />
    <AdministeredQueue name="queue3" />
  </AdministeredDestinations>
 

import org.exolab.jms.administration.AdminConnectionFactory;
import org.exolab.jms.administration.JmsAdminServerIfc;

// ...
    String url = "tcp://localhost:3035/";
    JmsAdminServerIfc admin = AdminConnectionFactory.create(url);

    String queue = "myqueue";
    Boolean isQueue = Boolean.TRUE;
    if (!admin.addDestination(queue, isQueue)) {
        System.err.println("Failed to create queue " + queue);
    }
        

utworzenie destynacji-tematy o nazwie mytopic:

    String topic = "mytopic";
    Boolean isQueue = Boolean.FALSE;
    if (!admin.addDestination(topic, isQueue)) {
        System.err.println("Failed to create topic " + topic);
    }
        

4. Programowanie klientów JMS

      Context ctx = new InitialContext();
      ConnectionFactory fact = (ConnectionFactory) ctx.lookup("ConnectionFactory");
      String admDestName = "myqueue";
      Destination dest = (Destination) ctx.lookup(admDestName);

1. tutaj nazwą JNDI dla fabryki połączeń jest "ConnectionFactory", może być inna nazwa (np. jms/ConnectionFactory)
2. dostawca inicjalnego kontekstu musi być okreslony w jndi.properties albo za pomoca właściwości systemowych. zgodnie z konfiguracją serwera JMS.

      Connection con = fact.createConnection();
      Session ses = con.createSession(false, Session.AUTO_ACKNOWLEDGE);

// false - bez transakcyjnosci
// AUTO_ACKNOWLEDGE - automatyczne potwierdzanie

con.start();
// ....
// Nadawanie lub odbiór
//...
con.close();

Wiadomości tworzymy za pomocą odpowiednich metod intefejsu Session np:

// Utworzenie wiadomości tekstowej:
TextMessage msg = ses.createTextMessage();
msg.setText(...);

// Utworzenie nadawcy, który będzie posyłał wiadomości do destynacji dest
MessageProducer sender = ses.createProducer(dest);

// Posłanie wiadomości msg
sender.send(msg);

// Utworzenie odbiorcy, który będzie odbierał wiadomości z destynacji dest
MessageConsumer receiver = ses.createConsumer(dest);

// Odbiór
Message msg = receiver.receive(); // blokowanie,
                                  //dopóki wiadomośc nie będzie dostępna
// ....
Message msg = receiver.receive(n); // blokowanie (ale nie dłużej niż n ms)
// ...
Message msg = receiver.receiveNoWait(); // wraca od razu; null - brak wiadomości

class MsgListener implements MessageListener {
   public void onMessage(Message msg) {
       // ... przetwarzanie wiadomości msg
   }
}

// ....
MessageConsumer receiver;
// .... 
MessageListener msgl = new MsgListener();
//....
receiver.setMessageListener(msgl);

5. Komunikaty w trybie point-to-point

import javax.naming.*;
import javax.jms.*;

public class Sender {

  public static void main(String[] args) {

    Connection con = null;
    try {
      Context ctx = new InitialContext();
      ConnectionFactory factory = (ConnectionFactory) ctx.lookup("ConnectionFactory");
      String admDestName = args[0];
      Destination dest = (Destination) ctx.lookup(admDestName);
      con = factory.createConnection();
      Session ses = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageProducer sender = ses.createProducer(dest);
      con.start();
      TextMessage msg = ses.createTextMessage();
      msg.setText(args[1]);
      sender.send(msg);
      System.out.println("Sender sent msg: " + args[1]);
    } catch (Exception exc) {
        exc.printStackTrace();
        System.exit(1);
    } finally {
        if (con != null) {
          try {
            con.close();
          } catch (JMSException exc) {
              System.err.println(exc);
          }
        }
    }
    System.exit(0);
  }
}

import javax.naming.*;
import javax.jms.*;

public class Receiver {

  public static void main(String[] args) {
      Connection con = null;
      try {
        Context ctx = new InitialContext();
        ConnectionFactory factory = (ConnectionFactory) ctx.lookup("ConnectionFactory");
        String admDestName = args[0];
        Destination dest = (Destination) ctx.lookup(admDestName);
        con = factory.createConnection();
        Session ses = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
        MessageConsumer receiver = ses.createConsumer(dest);

        con.start();
        System.out.println("Receiver started");
        Message msg = receiver.receive();
        if (msg instanceof TextMessage) {
            TextMessage text = (TextMessage) msg;
            System.out.println("Received: " + text.getText());
        } else if (msg != null) {
            System.out.println("Received non text message");
        }
      } catch (Exception exc) {
          exc.printStackTrace();
          System.exit(1);
      } finally {
          if (con != null) {
            try {
              con.close();
            } catch (JMSException exc) {
                System.err.println(exc);
            }
          }
      }
      System.exit(0);
  }
}

Zobacz prezentację multimedialną  

6. Słuchacze wiadomości - przykład

import javax.naming.*;
import javax.jms.*;
import javax.swing.*;
import java.awt.event.*;

import javax.naming.*;
import javax.jms.*;
import javax.swing.*;
import java.awt.event.*;

public class AsynchrReceiver extends JFrame implements MessageListener {

  private Connection con;
  private JTextArea ta = new JTextArea(10, 20);

  public AsynchrReceiver(String destName) {

    try {
      Context ctx = new InitialContext();
      ConnectionFactory factory = (ConnectionFactory) ctx.lookup("ConnectionFactory");
      Destination dest = (Destination) ctx.lookup(destName);
      con = factory.createConnection();
      Session ses = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer receiver = ses.createConsumer(dest);
      receiver.setMessageListener(this);
      con.start();
    } catch (Exception exc) {
         exc.printStackTrace();
         System.exit(1);
    }

    add(new JScrollPane(ta));

    addWindowListener(new WindowAdapter() {
      public void windowClosing(WindowEvent e) {
        try { con.close(); } catch(Exception exc) {}
        dispose();
        System.exit(0);
      }
    });
    setTitle("Czekam");
    pack();
    setLocationRelativeTo(null);
    show();
  }

  int i=0;
  public void onMessage(Message msg) {
    setTitle("Received msg " + ++i);
    try {
      ta.append(((TextMessage) msg).getText() + "\n");
    } catch(JMSException exc) { System.err.println(exc); }
  }

  public static void main(String[] args) {
    new AsynchrReceiver("queue1");
  }

}

Zobacz przentację mulimedialną działania programu

7. Działanie w trybie publisher/subscriber

8. Trwali subskrybenci

String subName = "MySub";
MessageConsumer topicSubscriber =
  session.createDurableSubscriber(myTopic, subName);

import javax.naming.*;
import javax.jms.*;

public class DurableSubscriber {

  static void printMsg(Message message) throws Exception {
    if (message instanceof TextMessage) {
        TextMessage text = (TextMessage) message;
        System.out.println("Received: " + text.getText());
    } else if (message != null) {
        System.out.println("Received non text message");
    }
  }

  public static void main(String[] args) {
    Connection con = null;
    try {
      Context ctx = new InitialContext();
      ConnectionFactory factory = (ConnectionFactory) ctx.lookup("ConnectionFactory");
      String admTopicName = args[0];
      String subscriptionName = args[1];
      Topic topic = (Topic) ctx.lookup(admTopicName);
      con = factory.createConnection();
      Session ses = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
      TopicSubscriber subs = ses.createDurableSubscriber(topic, subscriptionName);

      con.start();
      while (true) {
        Message message = subs.receiveNoWait();
        if (message == null) {
          System.out.println("Waiting... Press ctrl-c to stop");
          try { Thread.sleep(10000); } catch (Exception exc) { break; }
        }
        else printMsg(message);
      }
    } catch (Exception exc) {
      exc.printStackTrace();
      System.exit(1);
    } finally {
      if (con != null) {
          try {
              con.close();
          } catch (JMSException exc) {
              System.err.println(exc);
          }
      }
    }
    System.exit(0);
  }

}

1. jeżeli potwierdzamy odbiór wiadomości (a tak jest w trybie AUTOMATIC_ACKNOWLEDGE), to otrzymane wiadomości nie będą ponownie posyłane
2. możemy ustalić czas przechowywania nieodebranych wiadomości (domyślnie wiadomości zachowane są na stale).

9. Podsumowanie

• w trybie point-to-point,
• asynchronicznie, poprzez nasłuch komunikatów,
• w trybie publikacji-subskrypcji.

10. Zadania