JMS之ActiveMQ--发布-订阅消息模式（一对多）实现

喜欢ヅ旅行 2022-05-19 20:34 268阅读 0赞

## 1.创建消息发布者 ##

/** 
 * Title: MessageProducer.java 
 * Description: 
 * @author lihongjie 
 * @date 2018年7月15日 
 * @version 1.0 
 */ 
 package com.zit.mq2;
 
 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;
 
 /** 
 * Title: MessageProducer 
 * Description:消息生产者--消息发布者 
 * @author lihongjie 
 * @date 2018年7月15日 
 */
 public class MyMessageProducer { 
 	// 默认的连接用户名
 	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; 
 		//连接
 		Connection connection = null;
 		
 		Session session ;// 会话 接受或者发送消息的线程
 		Destination destination ;//消息的目的地
 		MessageProducer messageProducer ;//消息生产者
 		
 		//实例化连接工厂
 		connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL);
 		
 		try {
 			//通过连接工厂获取连接
 			connection = connectionFactory.createConnection();
 			//启动连接
 			connection.start();
 			
 			session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
 			//创建消息订阅者 destination = session.createTopic("Topic1"); //创建消息生产者
 			messageProducer = session.createProducer(destination);
 			//发送消息
 			sendMessage(session, messageProducer);
 			session.commit();
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 		//关闭资源
 		finally {
 			if(null != connection)
 				try {
 					connection.close();
 				} catch (JMSException e) {
 					e.printStackTrace();
 				}
 		}
 		
 		
 	}
 	/**
 	 * 
 	 * Title: sendMessage 
 	 * Description: 发送消息 
 	 * @param session
 	 * @param messageProducer
 	 */
 	public static void sendMessage(Session session, MessageProducer messageProducer) {
 		try {
 			for (int i = 0; i < SENDNUM; i++) {
 				TextMessage textMessage = session.createTextMessage("ActiveMQ发布的第"+i+"条消息！");
 				System.out.println("发布消息"+i+"条！");
 				messageProducer.send(textMessage);
 			}
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 	}
 }

## 2.创建监听 ##

### 2.1创建订阅者1的监听 ###

/** 
 * Title: MyListener.java 
 * Description: 
 * @author lihongjie 
 * @date 2018年7月15日 
 * @version 1.0 
 */ 
 package com.zit.mq2;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageListener;
 import javax.jms.TextMessage;
 
 /** 
 * Title: MyListener 
 * Description:消息监听--订阅者1 
 * @author lihongjie 
 * @date 2018年7月15日 
 */
 public class MyListener implements MessageListener {
 
 	/* (non-Javadoc) 
 	 * Title: onMessage 
 	 * Description: 
 	 * @param message 
 	 * @see javax.jms.MessageListener#onMessage(javax.jms.Message) 
 	 */
 	@Override
 	public void onMessage(Message message) {
 		
 		try {
 			System.out.println("通过监听，监听订阅者1收到的消息："+((TextMessage)message).getText());
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 
 	}
 
 }

### 2.2创建订阅者2的监听 ###

##  ##

/** 
 * Title: MyListener.java 
 * Description: 
 * @author lihongjie 
 * @date 2018年7月15日 
 * @version 1.0 
 */ 
 package com.zit.mq2;
 
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageListener;
 import javax.jms.TextMessage;
 
 /** 
 * Title: MyListener 
 * Description:消息监听--订阅者2 
 * @author lihongjie 
 * @date 2018年7月15日 
 */
 public class MyListener2 implements MessageListener {
 
 	/* (non-Javadoc) 
 	 * Title: onMessage 
 	 * Description: 
 	 * @param message 
 	 * @see javax.jms.MessageListener#onMessage(javax.jms.Message) 
 	 */
 	@Override
 	public void onMessage(Message message) {
 		
 		try {
 			System.out.println("通过监听，监听订阅者2收到的消息："+((TextMessage)message).getText());
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 
 	}
 
 }

##     ##

## 3.创建消息消费者--（订阅者） ##

### 3.1订阅者1 ###

/** 
 * Title: MessageProducer.java 
 * Description: 
 * @author lihongjie 
 * @date 2018年7月15日 
 * @version 1.0 
 */ 
 package com.zit.mq2;
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
 import javax.jms.JMSException;
 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;
 
 /** 
 * Title: MessageProducer 
 * Description:消息消费者 --订阅者1 
 * @author lihongjie 
 * @date 2018年7月15日 
 */
 public class MyMessageConsumer { 
 	// 默认的连接用户名
 	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; 
 		//连接
 		Connection connection = null;
 		Session session ;// 会话 接受或者发送消息的线程
 		Destination destination ;//消息的目的地
 		MessageConsumer messageConsumer;//消息消费者
 		
 		//实例化连接工厂
 		connectionFactory = new ActiveMQConnectionFactory(MyMessageConsumer.USERNAME, MyMessageConsumer.PASSWORD, MyMessageConsumer.BROKEURL);
 		
 		try {
 			//通过连接工厂获取连接
 			connection = connectionFactory.createConnection();
 			//启动连接
 			connection.start();
 			//创建session
 			session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
 			//创建消息订阅者
 			destination = session.createTopic("Topic1");
 			//创建消息消费者
 			messageConsumer = session.createConsumer(destination);
 			//接收消息--注册消息监听
 			messageConsumer.setMessageListener(new MyListener());
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 		
 	}
 }

### 3.2订阅者2 ###

/** 
 * Title: MessageProducer.java 
 * Description: 
 * @author lihongjie 
 * @date 2018年7月15日 
 * @version 1.0 
 */ 
 package com.zit.mq2;
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
 import javax.jms.Destination;
 import javax.jms.JMSException;
 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;
 
 /** 
 * Title: MessageProducer 
 * Description:消息消费者 --优化 
 * @author lihongjie 
 * @date 2018年7月15日 
 */
 public class MyMessageConsumer2 { 
 	// 默认的连接用户名
 	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; 
 		//连接
 		Connection connection = null;
 		Session session ;// 会话 接受或者发送消息的线程
 		Destination destination ;//消息的目的地
 		MessageConsumer messageConsumer;//消息消费者
 		
 		//实例化连接工厂
 		connectionFactory = new ActiveMQConnectionFactory(MyMessageConsumer2.USERNAME, MyMessageConsumer2.PASSWORD, MyMessageConsumer2.BROKEURL);
 		
 		try {
 			//通过连接工厂获取连接
 			connection = connectionFactory.createConnection();
 			//启动连接
 			connection.start();
 			//创建session
 			session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
 			//创建消息订阅者
 			destination = session.createTopic("Topic1");
 			//创建消息消费者
 			messageConsumer = session.createConsumer(destination);
 			//接收消息--注册消息监听
 			messageConsumer.setMessageListener(new MyListener2());
 			/*TextMessage textMessage = (TextMessage)messageConsumer.receive(100000);
 			System.out.println("第二个消费者，收到的消息是："+textMessage);*/
 		} catch (JMSException e) {
 			e.printStackTrace();
 		}
 		
 	}
 }

### 4. 查看消息队列页面 ###

![70][]

java工程代码下载地址：[点击前往下载][Link 1]

[70]: /images/20220520/6f8ed9f742af45fcb85be0da04b50710.png
[Link 1]: https://download.csdn.net/download/qq_37306041/10542370