JMS pub/sub 接口

TopicConnectionFactory是从 JNDI 中检索的受管理对象，它创建了到提供者的链接。它包含 createTopicConnection() 方法，该方法返回了一个 TopicConnection 对象。

TopicConnection 封装了到提供者的一个活动连接。它的一些方法包括：
createTopicSession(boolean, int)：返回 TopicSession 对象。boolean 参数指出 TopicSession 是否被处理，int 指出确认模式。
start() （从 Connection 继承）：激活提供者发送的消息。
stop() （从 Connection 继承）：临时停止发送消息，可以用 start() 重新开始发送。
close() （从 Connection 继承）：关闭到提供者的连接，并释放它所占有的所有资源。


TopicSession是发送和接收 pub/sub 消息的单线程上下文。它的一些方法包括：

createPublisher(Topic)：返回 TopicPublisher 对象来向指定的 Topic 发送消息。


createSubscriber(Topic)：返回 TopicSubscriber 对象来接收指定 Topic 中的消息。这个订阅者是 非持久的，也就是说，订阅只持续到对象的生存周期，只在它对象处于活动状态时才接收消息。


createDurableSubscriber(Topic, String)：返回 TopicSubscriber 对象来接收指定 Topic 中的消息，String 给出订阅者的名字。如果对象没有处于活动状态，则通过 JMS 保留 持久的 订阅者的消息，并将它发送给以后用相同名称创建的订阅者对象。


unsubscribe(String)：结束名为 String 的订阅者的订阅。


commit() （从 Session 继承）：提交当前事务的所有已使用或者已产生的消息。


rollback() （从 Session 继承）：回滚当前事务的所有已使用或者已产生的消息。


create<MessageType>Message(...) （从 Session 继承）：各种返回 <MessageType>Message 的方法，如 MapMessage、TextMessage 等。

Topic 封装了一个 pub/sub 目的地。它是从 JNDI 中检索的受管理对象。

TopicPublisher 被用来发送 pub/sub 消息。它的一些方法包括：

publish(Message)：发布指示的 Message。


setDeliveryMode(int) （从 MessageProducer 继承）：设置后续消息发送的发送模式，有效值为 DeliveryMode.PERSISTENT 和 DeliveryMode.NON_PERSISTENT。


setPriority(int) （从 MessageProducer 继承）：设置后续消息发送的优先级，有效值为 0 到 9。


setTimeToLive(long) （从 MessageProducer 继承）：设置后续发送消息失效前的持续时间，以毫秒为单位。


TopicSubscriber 被用来接收点对点消息。它的一些方法包括：

receive() （从 MessageConsumer 继承）：返回到达的下一个消息，这个方法会阻塞，直到一个消息可用为止。


receive(long) （从 MessageConsumer 继承）：接收在 long 毫秒内到达的下一个消息，如果在时间期限内没有消息到达，则返回 null 。


receiveNoWait （从 MessageConsumer 继承）：返回立即可用的下一个消息，如果没有消息可用，则返回 null。


setMessageListener(MessageListener) (inherited from MessageConsumer)：设置 MessageListener，MessageListener 在消息到达时接收消息，即异步接收消息。

TPublisher.java 的代码清单
import java.io.*;
import javax.jms.*;
import javax.naming.*;

public class TPublisher {

    public static void main(String[] args) {

        new TPublisher().publish();
    }
    
    public void publish() {

        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));

        try {
            //Prompt for JNDI names
            System.out.println("Enter TopicConnectionFactory name:");
            String factoryName = reader.readLine();
            System.out.println("Enter Topic name:");
            String topicName = reader.readLine();

            //Look up administered objects
            InitialContext initContext = new InitialContext();
            TopicConnectionFactory factory =
                (TopicConnectionFactory) initContext.lookup(factoryName);
            Topic topic = (Topic) initContext.lookup(topicName);
            initContext.close();

            //Create JMS objects
            TopicConnection connection = factory.createTopicConnection();
            TopicSession session =
                connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            TopicPublisher publisher = session.createPublisher(topic);

            //Send messages
            String messageText = null;
            while (true) {
                System.out.println("Enter message to send or 'quit':");
                messageText = reader.readLine();
                if ("quit".equals(messageText))
                    break;
                TextMessage message = session.createTextMessage(messageText);
                publisher.publish(message);
            }

            //Exit
            System.out.println("Exiting...");
            reader.close();
            connection.close();
            System.out.println("Goodbye!");

        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
}

TSubscriber.java 的代码清单
import java.io.*;
import javax.jms.*;
import javax.naming.*;

public class TSubscriber implements MessageListener {

    private boolean stop = false;

    public static void main(String[] args) {

        new TSubscriber().subscribe();
    }

    public void subscribe() {

        BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));

        try {
            //Prompt for JNDI names
            System.out.println("Enter TopicConnectionFactory name:");
            String factoryName = reader.readLine();
            System.out.println("Enter Topic name:");
            String topicName = reader.readLine();
            reader.close();

            //Look up administered objects
            InitialContext initContext = new InitialContext();
            TopicConnectionFactory factory =
                (TopicConnectionFactory) initContext.lookup(factoryName);
            Topic topic = (Topic) initContext.lookup(topicName);
            initContext.close();

            //Create JMS objects
            TopicConnection connection = factory.createTopicConnection();
            TopicSession session =
                connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            TopicSubscriber subscriber = session.createSubscriber(topic);
            subscriber.setMessageListener(this);
            connection.start();

            //Wait for stop
            while (!stop) {
                Thread.sleep(1000);
            }

            //Exit
            System.out.println("Exiting...");
            connection.close();
            System.out.println("Goodbye!");

        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

    public void onMessage(Message message) {

        try {
            String msgText = ((TextMessage) message).getText();
            System.out.println(msgText);
            if ("stop".equals(msgText))
                stop = true;
        } catch (JMSException e) {
            e.printStackTrace();
            stop = true;
        }
    }
}

评论

2 楼 mxmredfox 2007-04-20  

這個程序有問題。。沒有初始化上下文。。

1 楼 mxmredfox 2007-04-20  

Need to specify class name in environment or system property, or as an applet parameter, or in an application resource file:  java.naming.factory.initialcn