jboss4官方jms程序(Pub-Sub)

两个类：

TopicRecvClient.java(接收类subscriber)

TopicSendClient.java(发送类publisher)

 

TopicRecvClient.java:

package com.rox.jms;

import java.util.Properties;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.Topic;

import javax.jms.TopicConnection;

import javax.jms.TopicConnectionFactory;

import javax.jms.TopicPublisher;

import javax.jms.TopicSubscriber;

import javax.jms.TopicSession;

import javax.jms.TextMessage;

import javax.naming.Context;
import javax.naming.InitialContext;

import javax.naming.NamingException;

/**
 * A JMS client example program that synchronously receives a message a Topic
 *
 *
 * @author [e-mail]
 *
 * @version $Revision:$
 *
 */

public class TopicRecvClient

{

    TopicConnection conn = null;

    TopicSession session = null;

    Topic topic = null;

    public void setupPubSub()

    throws JMSException, NamingException

    {
        Properties environment = new Properties();
        environment.put(Context.PROVIDER_URL, "localhost:1099");
        environment.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");
        InitialContext iniCtx = new InitialContext(environment);

        Object tmp = iniCtx.lookup("ConnectionFactory");

        TopicConnectionFactory tcf = (TopicConnectionFactory) tmp;

        conn = tcf.createTopicConnection();

        topic = (Topic) iniCtx.lookup("topic/testTopic");

        session = conn.createTopicSession(false,

        TopicSession.AUTO_ACKNOWLEDGE);

        conn.start();

    }

    public void recvSync()

    throws JMSException, NamingException

    {

        System.out.println("Begin recvSync");

        // Setup the pub/sub connection, session

        setupPubSub();

        // Wait upto 5 seconds for the message

        TopicSubscriber recv = session.createSubscriber(topic);

        Message msg = recv.receive(5000);

        if (msg == null)

            System.out.println("Timed out waiting for msg");

        else

            System.out.println("TopicSubscriber.recv, msgt=" + msg);

    }

    public void stop() throws JMSException

    {

        conn.stop();

        session.close();

        conn.close();

    }

    public static void main(String args[]) throws Exception

    {

        System.out.println("Begin TopicRecvClient, now="
                + System.currentTimeMillis());

        TopicRecvClient client = new TopicRecvClient();

        client.recvSync();

        client.stop();

        System.out.println("End TopicRecvClient");

        System.exit(0);

    }

}
TopicSendClient.java:

package com.rox.jms;

import java.util.Properties;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.Topic;

import javax.jms.TopicConnection;

import javax.jms.TopicConnectionFactory;

import javax.jms.TopicPublisher;

import javax.jms.TopicSubscriber;

import javax.jms.TopicSession;

import javax.jms.TextMessage;

import javax.naming.Context;
import javax.naming.InitialContext;

import javax.naming.NamingException;

/**
 * A JMS client example program that sends a TextMessage to a Topic
 *
 *
 * @author [e-mail]
 *
 * @version $Revision:$
 *
 */

public class TopicSendClient

{

    TopicConnection conn = null;

    TopicSession session = null;

    Topic topic = null;

    public void setupPubSub()

    throws JMSException, NamingException

    {
        Properties environment = new Properties();
        environment.put(Context.PROVIDER_URL, "localhost:1099");
        environment.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");
        InitialContext iniCtx = new InitialContext(environment);

        Object tmp = iniCtx.lookup("ConnectionFactory");

        TopicConnectionFactory tcf = (TopicConnectionFactory) tmp;

        conn = tcf.createTopicConnection();

        topic = (Topic) iniCtx.lookup("topic/testTopic");

        session = conn.createTopicSession(false,

        TopicSession.AUTO_ACKNOWLEDGE);

        conn.start();

    }

    public void sendAsync(String text)

    throws JMSException, NamingException

    {

        System.out.println("Begin sendAsync");

        // Setup the pub/sub connection, session

        setupPubSub();

        // Send a text msg

        TopicPublisher send = session.createPublisher(topic);

        TextMessage tm = session.createTextMessage(text);

        send.publish(tm);

        System.out.println("sendAsync, sent text="

        + tm.getText());

        send.close();

        System.out.println("End sendAsync");

    }

    public void stop() throws JMSException

    {

        conn.stop();

        session.close();

        conn.close();

    }

    public static void main(String args[]) throws Exception

    {

        System.out.println("Begin TopicSendClient, now="
                + System.currentTimeMillis());

        TopicSendClient client = new TopicSendClient();

        client.sendAsync("A text msg, now=" + System.currentTimeMillis());

        client.stop();

        System.out.println("End TopicSendClient");

        System.exit(0);

    }

}
　　运行的时候先运行TopicRecvClient,然后迅速(5000毫秒之内)运行TopicSendClient,会看到TopicRecvClient的控制台中出现接收的信息．topic的消息只能在发送的时候接收到，如果再发送之后再去接收是收不到任何信息的．

　　还有一种介于p2p和pub2sub之间的一种jms类型(durable pub2sub),写一个java类：

DurableTopicRecvClient.java:

package com.rox.jms;

import java.util.Properties;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.Topic;

import javax.jms.TopicConnection;

import javax.jms.TopicConnectionFactory;

import javax.jms.TopicPublisher;

import javax.jms.TopicSubscriber;

import javax.jms.TopicSession;

import javax.jms.TextMessage;

import javax.naming.Context;
import javax.naming.InitialContext;

import javax.naming.NamingException;

/**
 * A JMS client example program that synchronously receives a message a Topic
 *
 *
 *
 * @author [e-mail]
 *
 * @version $Revision:$
 *
 */

public class DurableTopicRecvClient

{

    TopicConnection conn = null;

    TopicSession session = null;

    Topic topic = null;

    public void setupPubSub()

    throws JMSException, NamingException

    {
        Properties environment = new Properties();
        environment.put(Context.PROVIDER_URL, "localhost:1099");
        environment.put(Context.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");
        InitialContext iniCtx = new InitialContext(environment);

        Object tmp = iniCtx.lookup("ConnectionFactory");

        TopicConnectionFactory tcf = (TopicConnectionFactory) tmp;

        conn = tcf.createTopicConnection("john", "needle");

        topic = (Topic) iniCtx.lookup("topic/testTopic");

        session = conn.createTopicSession(false,

        TopicSession.AUTO_ACKNOWLEDGE);

        conn.start();

    }

    public void recvSync()

    throws JMSException, NamingException

    {

        System.out.println("Begin recvSync");

        // Setup the pub/sub connection, session

        setupPubSub();

        // Wait upto 5 seconds for the message

        TopicSubscriber recv = session.createDurableSubscriber(topic,
                "chap6-ex1dtps");

        Message msg = recv.receive(5000);

        if (msg == null)

            System.out.println("Timed out waiting for msg");

        else

            System.out.println("DurableTopicRecvClient.recv, msgt=" + msg);

    }

    public void stop() throws JMSException

    {

        conn.stop();

        session.close();

        conn.close();

    }

    public static void main(String args[]) throws Exception

    {

        System.out.println("Begin DurableTopicRecvClient, now="
                + System.currentTimeMillis());

        DurableTopicRecvClient client = new DurableTopicRecvClient();

        client.recvSync();

        client.stop();

        System.out.println("End DurableTopicRecvClient");
        System.exit(0);
    }

}　
　　这时候就可以先运行TopicSendClient了，然后再运行DurableTopicRecvClient,也可以收到消息．

     DurableTopicRecvClient.java中有两句代码需要引起注意:

     conn = tcf.createTopicConnection("john", "needle");

     TopicSubscriber recv = session.createDurableSubscriber(topic,"chap6-ex1dtps");

    

评论

1 楼 卒子99 2007-04-20  

不知道你分析过这段代码吗了？我用的是jboss4.0，上面的例子也已经调通了，不过还是有一个问题，就是p/s这种模式下使用Durable订阅，在书上看到是需要指定用户名和密码，就是下面这段代码：
conn = tcf.createTopicConnection("john", "needle");

之前也查阅过一些文章，是要把example/jms下的jbossmq-state.xml复制到服务器的conf下。我已经试过了，如果不部署这个配置文件已经能订阅成功。但是我在配置文件中修改了john的密码，却没有任何作用。所以我想请问，这个jms用户列表是在哪儿定义的呢？