JMS(二)之发布/订阅模式

发布/订阅模式（Publish/Subscribe）。发布/订阅消息系统支持一个事件驱动模型，消息生产者和消费者都参与消息的传递
发布/订阅模式 必须是先订阅，后发布 

举个例子：

第一：先创建一个消息订阅者

package com.activemq.publish_subscribe;

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;

/**

 * 必须是先订阅，后发布 

 * 消息订阅者一

 * */

public class JMSConsumer1 {

private static final String USERNAME=ActiveMQConnection.DEFAULT_USER;//默认的连接名称
private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;//默认密码
private static final String BROKERURL=ActiveMQConnection.DEFAULT_BROKER_URL;//默认的连接地址

public static void main(String[] args) {
ConnectionFactory connectionFactory;//连接工厂
Connection connection = null;//连接
Session session; //会话 接受或发送消息的线程
Destination destination;//消息的目的地
MessageConsumer messageConsumer;//消息消费者
//实例化连接工厂  默认连接地址：tcp://localhost:61616
connectionFactory = new ActiveMQConnectionFactory(JMSConsumer1.USERNAME, JMSConsumer1.PASSWORD, JMSConsumer1.BROKERURL);
try {
connection=connectionFactory.createConnection();//通过连接工厂连接
connection.start();//启动连接
session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

            //必须与JMSProducer中的生产者名称相同
destination=session.createTopic("FirstTopic1");
messageConsumer=session.createConsumer(destination);//创建消息消费者
messageConsumer.setMessageListener(new Listener1());//注册消息监听

} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

}

然后创建一个消费生产者

package com.activemq.publish_subscribe;

import javax.jms.Connection;

import javax.jms.ConnectionFactory;

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 JMSProducer {
private static final String USERNAME=ActiveMQConnection.DEFAULT_USER;//默认的连接名称
private static final String PASSWORD=ActiveMQConnection.DEFAULT_PASSWORD;//默认连接密码
private static final String BROKERURL=ActiveMQConnection.DEFAULT_BROKER_URL;//默认的连接地址
private static final int SENDNUM=10;//发送消息的数量

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

ConnectionFactory connectionFactory;//连接工厂
Connection connection = null;//连接
Session session; //会话 接受或发送消息的线程
Destination destination;//消息的目的地
MessageProducer messageProducer;//消息生产者
//实例化连接工厂  默认连接地址：tcp://localhost:61616
connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKERURL);
try {
connection = connectionFactory.createConnection();//通过连接工厂获取连接
connection.start();//启动连接
  
   session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);//创建session会话
   destination=session.createTopic("FirstTopic1");//生产者名称
   messageProducer=session.createProducer(destination);//创建消息生产者
  sendMessage(session, messageProducer);//发送消息
  session.commit();//提交事务
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}finally{
if(connection!=null){
connection.close();//关闭
}
}

}
/**
* 发送消息
* @throws Exception 
* */
private static void sendMessage(Session session,MessageProducer messageProducer) throws Exception{
for(int i=0;i<JMSProducer.SENDNUM;i++){
TextMessage message=session.createTextMessage("ActiveMQ 发送的消息"+i);
   System.out.println("发送消息"+"ActiveMQ发布的消息"+i);

            messageProducer.send(message);//设置一条一条的发送
}

}

}

创建完成以后，给它创建一个监听

package com.activemq.publish_subscribe;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

/**

 * 消息监听-订阅者一

 * */

public class Listener1 implements MessageListener {

@Override
public void onMessage(Message message) {
// TODO Auto-generated method stub
try {
System.out.println("订阅者一收到的消息："+((TextMessage)message).getText());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

}



大家可以亲自尝试一下，也许会有不一样的收获