JMS发布订阅模式

方式一：
package com.deppon.test04.jms.topicpublish;
发布者：
import java.util.Properties;

import javax.jms.DeliveryMode; 
import javax.jms.Message; 
import javax.jms.MessageConsumer; 
import javax.jms.Session; 
import javax.jms.Topic; 
import javax.jms.TopicConnection; 
import javax.jms.TopicConnectionFactory; 
import javax.jms.TopicPublisher; 
import javax.jms.TopicSession; 
import javax.jms.TopicSubscriber; 
import javax.naming.Context; 
import javax.naming.InitialContext; 
 
public class SimplePublisher { 
 
    private TopicPublisher producer; 
    private TopicSession session; 
    private TopicConnection connection; 
    private boolean isOpen = true; 
     
    public SimplePublisher() throws Exception{ 
       
        Properties props = new Properties();
//        props.setProperty(Context.INITIAL_CONTEXT_FACTORY,"org.apache.activemq.jndi.ActiveMQInitialContextFactory");
//        props.setProperty(Context.PROVIDER_URL,"tcp://localhost:61616");
        props.setProperty("connectionFactoryNames","TopicCF");
        props.setProperty("queue.queue1","jms.queue1");
        props.setProperty("topic.topic1","jms.topic1");
        javax.naming.Context context = new InitialContext(props);
       
//        Context context = new InitialContext(); 
        TopicConnectionFactory connectionFactory = (TopicConnectionFactory)context.lookup("TopicCF"); 
        connection = connectionFactory.createTopicConnection(); 
        connection.setClientID("OK111"); 
        session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); 
        Topic topic = (Topic)context.lookup("topic1"); 
        producer = session.createPublisher(topic);//non durable 
        producer.setDeliveryMode(DeliveryMode.PERSISTENT); 
        connection.start(); 
         
    } 
     
     
    public boolean send(String text) { 
        if(!isOpen){ 
            throw new RuntimeException("session has been closed!"); 
        } 
        try{ 
            Message message = session.createTextMessage(text); 
            producer.send(message); 
            return true; 
        }catch(Exception e){ 
            return false; 
        } 
    } 
     
    public synchronized void close(){ 
        try{ 
            if(isOpen){ 
                isOpen = false; 
            } 
            session.close(); 
            connection.close(); 
        }catch (Exception e) { 
            // 
        } 
    } 
     
} 

订阅者：
package com.deppon.test04.jms.topicpublish;

import java.util.Properties;

import javax.jms.Session; 
import javax.jms.Topic; 
import javax.jms.TopicConnection; 
import javax.jms.TopicConnectionFactory; 
import javax.jms.TopicSession; 
import javax.jms.TopicSubscriber; 
import javax.naming.Context; 
import javax.naming.InitialContext; 
   
public class SimpleSubscriber { 
 
    private TopicConnection connection; 
    private TopicSession session; 
    private TopicSubscriber consumer; 
     
    private boolean isStarted; 
     
    public SimpleSubscriber(String clientId) throws Exception{ 
       
        Properties props = new Properties();
        props.setProperty(Context.INITIAL_CONTEXT_FACTORY,"org.apache.activemq.jndi.ActiveMQInitialContextFactory");
        props.setProperty(Context.PROVIDER_URL,"tcp://localhost:61616");
        props.setProperty("connectionFactoryNames","TopicCF");
        props.setProperty("queue.queue1","jms.queue1");
        props.setProperty("topic.topic1","jms.topic1");
        javax.naming.Context context = new InitialContext(props);
       
//        Context context = new InitialContext(); 
        TopicConnectionFactory connectionFactory = (TopicConnectionFactory)context.lookup("TopicCF");
        connection = connectionFactory.createTopicConnection(); 
        connection.setClientID(clientId); 
        session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE); 
        Topic topic = (Topic)context.lookup("topic1"); 
        consumer = session.createDurableSubscriber(topic, "Test-subscriber"); 
        consumer.setMessageListener(new TopicMessageListener()); 
    } 
     
     
    public synchronized boolean start(){ 
        if(isStarted){ 
            return true; 
        } 
        try{ 
            connection.start(); 
            isStarted = true; 
            return true; 
       }catch(Exception e){ 
            return false; 
        } 
    } 
     
    public synchronized void close(){ 
        isStarted = false; 
        try{ 
            session.close(); 
            connection.close(); 
        }catch(Exception e){ 
            // 
        } 
    } 
} 

监听消息类：
package com.deppon.test04.jms.topicpublish;

import javax.jms.Message;
import javax.jms.MessageListener;

public class TopicMessageListener implements MessageListener {

    MessageListener paramMessageListener;

    public MessageListener getParamMessageListener() {
        return paramMessageListener;
    }

    public void setParamMessageListener(MessageListener paramMessageListener) {
        this.paramMessageListener = paramMessageListener;
    }

    @Override
    public void onMessage(Message paramMessage) {
//        this.paramMessageListener = paramMessage.get;
        // TODO Auto-generated method stub
        System.out.println("AAAAAAAAA BBBBB");
        System.out.println("AAAAAAAAA CCCCC");
    }
   
}
测试类：
package com.deppon.test04.jms.topicpublish;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class SimpleTestMain { 
      /**
     * @param args
     */ 
    public static void main(String[] args) throws Exception{ 
               
        SimpleSubscriber consumer = new SimpleSubscriber("TestClientId");
       
//        consumer.close();
        consumer.start(); 
         
        SimplePublisher productor = new SimplePublisher(); 
       
        for(int i=0; i<10; i++){ 
            productor.send("message content:" + i); 
        } 
        productor.close(); 
        //consumer.close(); 
    } 
     
 
} 

方式二：

发布者：
package com.deppon.test04.jms.topicpublish;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class SendTopic {
    private static final int SEND_NUMBER = 5;
    public static void sendMessage(Session session, MessageProducer producer)
            throws Exception {
        for (int i = 1; i <=SEND_NUMBER; i++) {
            TextMessage message = session
                    .createTextMessage("ActiveMq发送的消息" + i);
            //发送消息到目的地方
            System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
            producer.send(message);
        }
    }
  
    public static void main(String[] args) {
        // ConnectionFactory：连接工厂，JMS用它创建连接
        ConnectionFactory connectionFactory;
        // Connection：JMS客户端到JMS Provider的连接
        Connection connection = null;
        // Session：一个发送或接收消息的线程
        Session session;
        // Destination：消息的目的地;消息发送给谁.
        Destination destination;
        // MessageProducer：消息发送者
        MessageProducer producer;
        // TextMessage message;
        //构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
        connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                "tcp://localhost:61616");
        try {
            //构造从工厂得到连接对象
            connection = connectionFactory.createConnection();
            //启动
            connection.start();
            //获取操作连接
            session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
            //获取session注意参数值FirstTopic是一个服务器的topic(与queue消息的发送相比，这里是唯一的不同)
            destination = session.createTopic("FirstTopic");
            //得到消息生成者【发送者】
            producer = session.createProducer(destination);
            //设置不持久化，此处学习，实际根据项目决定
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            //构造消息，此处写死，项目就是参数，或者方法获取
            sendMessage(session, producer);
            session.commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection)
                    connection.close();
            } catch (Throwable ignore) {
            }
        }
    }
}

订阅者：
package com.deppon.test04.jms.topicpublish;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

public class ReceiveTopic implements Runnable {
    private String threadName;

    ReceiveTopic(String threadName) {
         this.threadName = threadName;
    }

    public void run() {
         // ConnectionFactory：连接工厂，JMS用它创建连接
         ConnectionFactory connectionFactory;
         // Connection：JMS客户端到JMS Provider的连接
         Connection connection =null;
         // Session：一个发送或接收消息的线程
         Session session;
         // Destination：消息的目的地;消息发送给谁.
         Destination destination;
         //消费者，消息接收者
         MessageConsumer consumer;
         connectionFactory = new ActiveMQConnectionFactory(
                    ActiveMQConnection.DEFAULT_USER,
                    ActiveMQConnection.DEFAULT_PASSWORD,"tcp://localhost:61616");
         try {
               //构造从工厂得到连接对象
               connection = connectionFactory.createConnection();
               //启动
               connection.start();
               //获取操作连接,默认自动向服务器发送接收成功的响应
               session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
               //获取session注意参数值FirstTopic是一个服务器的topic
               destination = session.createTopic("FirstTopic");
               consumer = session.createConsumer(destination);
               while (true) {
                    //设置接收者接收消息的时间，为了便于测试，这里设定为100s
                    TextMessage message = (TextMessage) consumer
                                .receive(100 * 1000);
                    if (null != message) {
                          System.out.println("线程"+threadName+"收到消息:" + message.getText());
                    } else {
                          continue;
                    }
               }
         } catch (Exception e) {
               e.printStackTrace();
         } finally {
               try {
                    if (null != connection)
                          connection.close();
               } catch (Throwable ignore) {
               }
         }
    }

    public static void main(String[] args) {
          //这里启动3个线程来监听FirstTopic的消息，与queue的方式不一样三个线程都能收到同样的消息
         ReceiveTopic receive1=new ReceiveTopic("thread1");
         ReceiveTopic receive2=new ReceiveTopic("thread2");
         ReceiveTopic receive3=new ReceiveTopic("thread3");
         Thread thread1=new Thread(receive1);
         Thread thread2=new Thread(receive2);
         Thread thread3=new Thread(receive3);
         thread1.start();
         thread2.start();
         thread3.start();
    }
}
参考：
http://activemq.apache.org/jndi-support.html
http://shift-alt-ctrl.iteye.com/blog/1915240
http://coderbase64.iteye.com/blog/2081937