パブリッシュ/サブスクライブはブロードキャストメカニズムの 1 つで、メッセージはトピックにパブリッシュされ、トピックのサブスクライバに自動的に配信されます。トピックは階層的に構成できます。こうすると最上位レベルのサブスクライバはすべてのメッセージを受信しますが、サブトピックのサブスクライバはサブトピックのメッセージだけを受信します。
パブリッシャは、次のオブジェクトを使用してメッセージをトピックに追加します。
javax.naming.Context
javax.jms.TopicConnectionFactory
javax.jms.TopicConnection
javax.jms.TopicSession
javax.jms.TopicPublisher
javax.jms.Message
パブリッシャクラスは、javax.jms
クラスのいずれも拡張しません。このクラスは明示的に他のクラスを拡張したり、暗黙的に java.lang.Object
を拡張したりできます。
次のコードを実行するには、JMC を使用して topic1 と名付けたトピックを定義する必要があります。
import java.util.*; import javax.jms.*; import javax.naming.*;
private Context _context = null; private TopicConnection _connection = null; private TopicSession _session = null; private TextMessage _message = null; private TopicPublisher _publisher = null;
// ユーザー ID とパスワード private String _user = new String("admin"); private String _password = new String("admin"); // トピックの名前を JMC 全体で一致させます。 // この情報は、SERVER-INF/jrun-resources.xml ファイルでも表示できます。 private String _topic = new String("topic1"); // これは、リモートホスト名とJNDI ポート番号を連結したものです。 private String _providerurl = new String("localhost:2918"); private String _ctxtFactory = new String ("jrun.naming.JRunContextFactory");
// JNDI で設定を行います。 try { Properties p = new Properties(); p.put(Context.SECURITY_PRINCIPAL, _user); p.put(Context.SECURITY_CREDENTIALS, _password); p.put(Context.PROVIDER_URL, _providerurl ); p.put( Context.INITIAL_CONTEXT_FACTORY, ctxtFactory); // 同じサーバ上のコンテキストにアクセスする場合は、 // これは空のコンストラクタでも構いません。 _context = new InitialContext(p); // Get TopicConnectionFactory from JNDI. final TopicConnectionFactory factory = (TopicConnectionFactory)_context.lookup ("jms/TopicConnectionFactory"); // ファクトリを使用して TopicConnection を作成します。 _connection = factory.createTopicConnection(); // TopicConnection を使用して TopicSession を作成します。 _session = _connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); // TopicSession を使用して TopicSender を作成します。 // topic1 を探します。 _publisher = _session.createPublisher((Topic)_context.lookup ("java:jms/topic/topic1")); // TextMessage オブジェクトを作成します。 _message = _session.createTextMessage(); } catch(NamingException e) { System.out.println("Naming Exception:" + e.getMessage()); } catch(JMSException e) { System.out.println("JMS Exception:" + e.getMessage()); }
int priority = Message.DEFAULT_PRIORITY; int delivery = Message.DEFAULT_DELIVERY_MODE; String text = new String("Blank Message"); String[] attrArray = req.getParameterValues("thisMessage"); // 呼び出しフォームには、thisMessage 用の値が 1 つしかないと想定します。 if(attrArray != null) { text = attrArray[0]; } try { System.out.println("setText の直前"); _message.setText(text); // トピックをパブリッシュします。 System.out.println("パブリッシュの直前" ); // トピックをパブリッシュします。メッセージは 5 分間存続します。 _publisher.publish(_message, delivery, priority, 5 * 60 * 1000); System.out.println("パブリッシュの直後"); } // try を終了 catch(JMSException e) { System.out.println("JMS Exception:" + e.getMessage()); }
// これらは if != null のチェックでラップしたほうがよいでしょう。 try { _publisher.close(); _session.close(); _connection.close(); _context.close(); } catch(NamingException e) { System.out.println("Naming exception in destroy: " + e.getMessage()); } catch(JMSException e) { System.out.println("JMS Exception:" + e.getMessage()); }
JMS では、関連するトピックに対してトピックがパブリッシュされると、これらがサブスクライバに渡されます。次のオブジェクトを使用して、サブスクライバをコーディングします。
javax.naming.Context
javax.jms.TopicConnectionFactory
javax.jms.TopicConnection
javax.jms.TopicSession
javax.jms.TopicSubscriber
javax.jms.Message
サブスクライバが非同期メッセージを使用する場合は、MessageListener
インターフェイスと onMessage
メソッドを実装する必要があります。サブスクライバクラスがサブスクライブしているトピックにメッセージがパブリッシュされると、JRun では onMessage
メソッドが呼び出されます。
サブスクライバクラスは、javax.jms
クラスのいずれも拡張しませんが、明示的に他のクラスを拡張したり、暗黙的に java.lang.Object
を拡張したりできます。
import java.util.*; import javax.jms.*; import javax.naming.*;
public class MyTopicSubscriber implements MessageListener {
private Context jndi = null; private TopicSession pubSession = null; private TopicSession subSession = null; private TopicPublisher publisher = null; private static TopicConnection connection = null; private String userName = null;
// トピックの名前を JMC 全体で一致させます。 // この情報は、SERVER-INF/jrun-resources.xml ファイルでも表示できます。 private String topicName = new String("topic1"); // これは、リモートホスト名とJNDI ポート番号を連結したものです。 private String providerurl = new String("localhost:2918"); private String factoryName = new String ("jrun.naming.JRunContextFactory"); Properties env = new Properties(); env.put(Context.PROVIDER_URL, providerurl ); env.put( Context.INITIAL_CONTEXT_FACTORY, "jrun.naming.JRunContextFactory" ); jndi = new InitialContext(env); TopicConnectionFactory conFactory = (TopicConnectionFactory)jndi.lookup(factoryName); connection=conFactory.createTopicConnection(); TopicSession pubSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); TopicSession subSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); // JMS トピックをルックアップします。 Topic thisTopic = (Topic)jndi.lookup(topicName); // JMS メッセージリスナを設定します。 TopicSubscriber subscriber=subSession.createSubscriber(thisTopic); subscriber.setMessageListener(this); // このサンプルはパブリッシュでも使用します。 publisher=pubSession.createPublisher(thisTopic); // アプリケーションを初期化します。 set(connection, pubSession, subSession, publisher, username); // JMS 接続を開始し、メッセージが配送されるようにします。 connection.start(); } catch(NamingException e) { System.out.println("Naming Exception:" + e.getMessage()); } catch(JMSException e) { System.out.println("JMS Exception:" + e.getMessage()); }
onMessage
メソッドをコーディングします。... public void onMessage(Message message) { try { TextMessage textMessage = (TextMessage) message; String text = textMessage.getText(); System.out.println(text); } catch(JMSException jmse){ jmse.printStackTrace(); } } ...