ActiveMQ消息队列的使用--点对点的消息模式

下载：apache-activemq-5.9.0-bin.zip

 目录结构介绍如下。

bin，放置启动、停止、注册服务等命令文件。

conf，配置文件，包括broker、jetty、log等配置文件。

data，放置消息缓存(kahadb)、jetty中jsp编译后的class、AMQ日志等文件。

docs，帮助文档

examples，应用示例

lib，依赖的jar包

webapps，AMQ下的web应用，包括web console 和 file server。

webapps-demo，demo应用，默认不能访问，如需访问需要在conf/jetty.xml文件中配置。

activemq-all-5.10.1.jar， AMQ的开发用的jar包

其它文件，包括license、notice和 readme

解压后apache-activemq-5.9.0\bin ，看到activemq.bat 启动服务

管理控制台地址：http://localhost:8161/admin/

账号admin

密码admin

可以实时看到发送端发送的消息，和接受端取消息

一：JMQ的两种消息模式

消息列队有两种消息模式，一种是点对点的消息模式，还有一种就是订阅的模式.

在MQ中，消息模型有两种，一种是队列（Queue），一种是主题（Topic）。队列是Point-To-Point的，队列中的消息，仅能被消费一次。主题是Pub/Sub模型，主题中的消息，可以由多个订阅者消费；订阅者只能消费它订阅以后的消息。这是遵循的JMS规范。

1.1:点对点的消息模式

点对点的模式主要建立在一个队列上面，当连接一个列队的时候，发送端不需要知道接收端是否正在接收，可以直接向ActiveMQ发送消息，发送的消息，将会先进入队列中，如果有接收端在监听，则会发向接收端，如果没有接收端接收，则会保存在activemq服务器，直到接收端接收消息，点对点的消息模式可以有多个发送端，多个接收端，但是一条消息，只会被一个接收端给接收到，哪个接收端先连上ActiveMQ，则会先接收到，而后来的接收端则接收不到那条消息

 

1.2：订阅模式

订阅/发布模式，同样可以有着多个发送端与多个接收端，但是接收端与发送端存在时间上的依赖，就是如果发送端发送消息的时候，接收端并没有监听消息，那么ActiveMQ将不会保存消息，将会认为消息已经发送，换一种说法，就是发送端发送消息的时候，接收端不在线，是接收不到消息的，哪怕以后监听消息，同样也是接收不到的。这个模式还有一个特点，那就是，发送端发送的消息，将会被所有的接收端给接收到，不类似点对点，一条消息只会被一个接收端给接收到。

 

二：点对点的实现代码

项目使用MAVEN来构建

    <dependencies>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-core</artifactId>
            <version>5.7.0</version>
        </dependency>
    </dependencies>

实例一：

2.1：点对点的发送端

package com.jiaoyinbi.activemq;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class PTPSend {
	// 默认连接用户名
	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
	// 默认连接密码
	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
	// 默认连接地址
	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
	// 发送的消息数量
	private static final int SENDNUM = 10;
	public void start() {


		// 连接工厂
		ConnectionFactory connectionFactory = null;
		// 连接
		Connection connection = null;
		// 一个操作会话
		Session session;
		// 消息目的地
		//目的地，其实就是连接到哪个队列，如果是点对点，那么它的实现是Queue，如果是订阅模式，那它的实现是Topic
		Destination destination;
		// 消息生产者，就是产生数据的对象
		MessageProducer messageProducer;
		// 实例化连接工厂
		try {
			// 根据用户名，密码，url实例化连接工厂
			connectionFactory = new ActiveMQConnectionFactory(PTPSend.USERNAME, PTPSend.PASSWORD, PTPSend.BROKEURL);
			// 通过连接工厂获得连接
			connection = connectionFactory.createConnection();
			// 启动连接
			connection.start();
			// 创建session
			
		   /**
			第一个参数：是否支持事务，如果为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
			第二个参数为false时，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
            Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
            Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。
            DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
            **/
			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
			// 创建一个名称为text-msg的消息队列
			destination = session.createQueue("text-msg");
			//从session中， 创建消息生产者
			messageProducer = session.createProducer(destination);


			//设置生产者得模式，有两种可选
			//DeliveryMode.PERSISTENT 当activemq关闭的时候，队列数据将会被保存
			//DeliveryMode.NON_PERSISTENT 当activemq关闭的时候，队列里面的数据将会被清空
			messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
			
			  //创建一条消息，当然，消息的类型有很多，如文字，字节，对象等,可以通过session.create..方法来创建出来
				// 创建一条文本消息
				TextMessage message = session.createTextMessage("呵呵。。" );
				// 通过消息生产者发出消息
				for (int i = 0; i < PTPSend.SENDNUM; i++) {
					// 发送消息
				     messageProducer.send(message);
			     }
				
				System.out.println("发送消息成功");
				////即便生产者的对象关闭了，程序还在运行哦
				session.commit();
				messageProducer.close();
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try {
					connection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}


	}
    public static void main(String[] args) {
        PTPSend send = new PTPSend();
        send.start();
    }
}

2.2：点对点的接收端

package com.jiaoyinbi.activemq;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class PTPReceive{
	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;// 默认连接用户名
	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;// 默认连接密码
	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;// 默认连接地址

	public void  start (){
		ConnectionFactory connectionFactory;// 连接工厂
		Connection connection = null;// 连接

		Session session;// 会话 接受或者发送消息的线程
		Destination destination;// 消息的目的地

		MessageConsumer messageConsumer;// 消息的消费者

		try {


			// 实例化连接工厂
			connectionFactory = new ActiveMQConnectionFactory(PTPReceive.USERNAME, PTPReceive.PASSWORD,
					PTPReceive.BROKEURL);
			// 通过连接工厂获取连接
			connection = connectionFactory.createConnection();
			// 启动连接
			connection.start();
			// 创建session
		    //第一个参数:是否支持事务，如果为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
            //第二个参数为false时，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
            //Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
            //Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。
            //DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			// 创建一个连接text-msg的消息队列
			destination = session.createQueue("text-msg");
			// 创建消息消费者,根据session，创建一个接收者对象
			messageConsumer = session.createConsumer(destination);
			   
            //实现一个消息的监听器
            //实现这个监听器后，以后只要有消息，就会通过这个监听器接收到
			messageConsumer.setMessageListener(new MessageListener() {
			
                public void onMessage(Message message) {
                    try {
                        //获取到接收的数据
                        String text = ((TextMessage)message).getText();
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
						
			
/*			while (true) {
				TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);
				if (textMessage != null) {
					System.out.println("收到的消息:" + textMessage.getText());
				} else {
					break;
				}
			}*/


		} catch (JMSException e) {
			e.printStackTrace();
		}


	}	
	 public static void main(String[] args) {
	        PTPReceive receive = new PTPReceive();
	        receive.start();
	    }
}

实例二：

2.3：点对点的发送端

package com.tgb.activemq;

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

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

/**
 * 消息生产者
 * 
 * @author zhaoshuhui
 *
 */
public class JMSProducer {
	// 默认连接用户名
	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
	// 默认连接密码
	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
	// 默认连接地址
	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
	// 发送的消息数量
	private static final int SENDNUM = 10;

	public static void main(String[] args) {
		// 连接工厂
		ConnectionFactory connectionFactory = null;
		// 连接
		Connection connection = null;
		// 会员接受或者发送消息的线程
		Session session;
		// 消息目的地
		Destination destination;
		// 消息生产者
		MessageProducer messageProducer;
		// 实例化连接工厂
		try {
			// 实例化连接工厂
			connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD,
					JMSProducer.BROKEURL);
			// 通过连接工厂获得连接
			connection = connectionFactory.createConnection();
			// 启动连接
			connection.start();
			// 创建session
			//http://localhost:8161/admin/
//			回复wanghongyu123：createSession(paramA,paramB);
//			paramA是设置事务的，paramB设置acknowledgment mode
//			paramA设置为false时：paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
//			paramA设置为true时：paramB的值忽略， acknowledgment mode被jms服务器设置为SESSION_TRANSACTED 。
//			Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。
//			Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会删除消息。
//			DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。在需要考虑资源使用时，这种模式非常有效。
			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
			// 创建一个名称为helloword的消息队列
			destination = session.createQueue("HelloWorld");
			// 创建消息生产者
			messageProducer = session.createProducer(destination);
			
			//发送消息
			for (int i = 0; i < JMSProducer.SENDNUM; i++) {
				// 创建一条文本消息
				TextMessage message = session.createTextMessage("ActiveMQ 发送消息" + i);
				 System.out.println("发送消息：Activemq 发送消息" + i);
				// 通过消息生产者发出消息
				messageProducer.send(message);
			}
			session.commit();

		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try {
					connection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 发送消息
	 * 
	 * @param session
	 * @param messageProducer
	 *            消息生产者
	 * @throws JMSException
	 */
	public static void sendMessage(Session session, MessageProducer messageProducer) throws JMSException {

		for (int i = 0; i < JMSProducer.SENDNUM; i++) {
			// 创建一条文本消息
			TextMessage message = session.createTextMessage("ActiveMQ 发送消息" + i);
			// 通过消息生产者发出消息
			messageProducer.send(message);
		}
	}
}

2.4：点对点的接收端

package com.tgb.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
//博客地址：http://blog.csdn.net/jiuqiyuliang/article/details/48608237
public class JMSConsumer {
	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;// 默认连接用户名
	private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;// 默认连接密码
	private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;// 默认连接地址

	public static void main(String[] args) {
		ConnectionFactory connectionFactory;// 连接工厂
		Connection connection = null;// 连接

		Session session;// 会话 接受或者发送消息的线程
		Destination destination;// 消息的目的地

		MessageConsumer messageConsumer;// 消息的消费者

		try {

			// 实例化连接工厂
			connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD,
					JMSConsumer.BROKEURL);
			// 通过连接工厂获取连接
			connection = connectionFactory.createConnection();
			// 启动连接
			connection.start();
			// 创建session
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			// 创建一个连接HelloWorld的消息队列
			destination = session.createQueue("HelloWorld");
			// 创建消息消费者
			messageConsumer = session.createConsumer(destination);

			while (true) {
				TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);
				if (textMessage != null) {
					System.out.println("收到的消息:" + textMessage.getText());
				} else {
					break;
				}
			}

		} catch (JMSException e) {
			e.printStackTrace();
		}

	}
}

：

二：订阅模式的实现代码

通过Topic发布和订阅消息

       为了使订阅者能够订阅消息，在运行程序时，需要先运行订阅者（Subscriber），后运行发布者（Publisher）。

实例一：

 1.通过Topic发布消息

package topic;

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

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

//发布者

public class Publisher {
	public static void main(String[] args) {
	    Publisher pb = new Publisher();
	    String msg = "Hello World!~~~~~";
	    pb.sendMessage(msg);
	    System.out.println("发送消息结束：" + msg);
	  }

	  /**
	   * 使用JMS向MQ发送消息
	   * @param msg 消息内容
	   */
	  public void sendMessage(String msg) {
	    // defualt user & password both are null
	    String user = ActiveMQConnection.DEFAULT_USER;
	    String password = ActiveMQConnection.DEFAULT_PASSWORD;
	    // DEFAULT_BROKER_URL =failover://tcp://localhost:61616
	    String url = ActiveMQConnection.DEFAULT_BROKER_URL;
	    String subject = "MQ.TOPIC";

	    // 1. 初始化连接工厂
	    ConnectionFactory contectionFactory = new ActiveMQConnectionFactory(user, password, url);
	    try {
	      // 2. 创建连接
	      Connection connection = contectionFactory.createConnection();
	      connection.start();
	      // 3.创建会话
	      Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
	      // 4. 创建要发布的主题，和Queue的区别就在此
	      Destination destination = session.createTopic(subject);
	      // 5. MessageProducer负责发送消息
	      MessageProducer producer = session.createProducer(destination);
	      TextMessage message = session.createTextMessage();
	      message.setStringProperty("hello", msg);
	      // 6. 发送消息
	      producer.send(message);
	      // 7. 关闭会话和连接
	      session.commit();
	      session.close();
	      connection.close();
	    } catch (JMSException e) {
	      e.printStackTrace();
	    }
	  }
}

2.通过Topic订阅消息

这里我们只有1个订阅者，想要验证多个订阅者，拷贝多份代码，改个类名即可。再次提醒，先运行订阅者

package topic;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
//订阅者
public class Subscriber {

	  public static void main(String[] args) {
	    String user = ActiveMQConnection.DEFAULT_USER;
	    String password = ActiveMQConnection.DEFAULT_PASSWORD;
	    String url = ActiveMQConnection.DEFAULT_BROKER_URL;
	    String subject = "MQ.TOPIC";
	    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
	    Connection connection;
	    try {
	      connection = connectionFactory.createConnection();
	      connection.start();
	      final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
	      Topic topic = session.createTopic(subject);
	      // MessageConsumer负责接受消息
	      MessageConsumer consumer = session.createConsumer(topic);
	      consumer.setMessageListener(new MessageListener() {

	        public void onMessage(Message msg) {
	          TextMessage message = (TextMessage) msg;
	          try {
	            String hello = message.getStringProperty("hello");
	            System.out.println("订阅者---SecondSubscriber---收到消息：\t" + hello);
	            session.commit();
	          } catch (JMSException e) {
	            e.printStackTrace();
	          }
	        }
	      });
	      // 为了测试效果，注释掉了两行代码，使Session和connection一直处于打开状态
	      //session.close();
	      //connection.close();
	    } catch (JMSException e) {
	      e.printStackTrace();
	    }
	  }
}

实例二：

先运行ReceiveTopic，再运行SendTopic 

1.发布端：

package com.jiaoyinbi.activemq;
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 main(String[] args) {
		//连接工厂，JMS用它创建连接
		ConnectionFactory connectionFactory;
		//JMS客户端到JMS Provider的连接
		Connection connection = null;
		//一个发送或接收消息的线程
		Session session;
		//消息的目的地;消息发送给谁.
		Destination destination;
		//消息发送者
		MessageProducer producer;
		// 构造ConnectionFactory实例对象，此处采用ActiveMq的实现jar
		connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
				ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
		try {
			// 构造从工厂得到连接对象
			connection = connectionFactory.createConnection();
			// 启动
			connection.start();
			// 获取操作连接
		    //第一个参数:是否支持事务，如果为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
            //第二个参数为false时，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
            //Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
            //Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。
            //DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
			
			
			// 获取session注意参数值FirstTopic是一个服务器的topic(与queue消息的发送相比，这里是唯一的不同)
			destination = session.createTopic("FirstTopic");
			
			
			// 得到消息生成者【发送者】
			producer = session.createProducer(destination);

			 //设置生产者的模式，有两种可选
            //设置不持久化，此处学习，实际根据项目决定  
            //DeliveryMode.PERSISTENT 当activemq关闭的时候，队列数据将会被保存
            //DeliveryMode.NON_PERSISTENT 当activemq关闭的时候，队列里面的数据将会被清空
			producer.setDeliveryMode(DeliveryMode.PERSISTENT);
			
			// 构造消息
			for (int i = 1; i <= SEND_NUMBER; i++) {
				TextMessage message = session.createTextMessage("ActiveMq发送的消息" + i);
				// 发送消息到目的地方
				System.out.println("发送消息：" + "ActiveMq 发送的消息" + i);
				producer.send(message);
			}
			session.commit();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (null != connection)
					connection.close();

			} catch (Throwable ignore) {

			}

		}

	}

}

2.订阅：

package com.jiaoyinbi.activemq;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
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;//线程名称
	
	
	//带参构造器
	public ReceiveTopic(String threadName) {
		this.threadName = threadName;
	}
	public void run() {
		ConnectionFactory connectionFactory;
		Connection connection = null;
		Session session;
		//消息的目的地;消息发送给谁.
		Destination destination;
		// 消费者，消息接收者
		MessageConsumer consumer;
		connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
		try {
			// 构造从工厂得到连接对象
			connection = connectionFactory.createConnection();
			// 启动
			connection.start();
			// 获取操作连接,默认自动向服务器发送接收成功的响应
		    //第一个参数:是否支持事务，如果为true，则会忽略第二个参数，被jms服务器设置为SESSION_TRANSACTED
            //第二个参数为false时，paramB的值可为Session.AUTO_ACKNOWLEDGE，Session.CLIENT_ACKNOWLEDGE，DUPS_OK_ACKNOWLEDGE其中一个。
            //Session.AUTO_ACKNOWLEDGE为自动确认，客户端发送和接收消息不需要做额外的工作。哪怕是接收端发生异常，也会被当作正常发送成功。
            //Session.CLIENT_ACKNOWLEDGE为客户端确认。客户端接收到消息后，必须调用javax.jms.Message的acknowledge方法。jms服务器才会当作发送成功，并删除消息。
            //DUPS_OK_ACKNOWLEDGE允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回，会话对象就会确认消息的接收；而且允许重复确认。
			session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			
			// 获取session注意参数值FirstTopic是一个服务器的topic
			destination = session.createTopic("FirstTopic");
			// 创建消息消费者,根据session，创建一个接收者对象
			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();
		
	}
}

订阅端打印：

控制台消息：