JMS之ActiveMQ基本应用

Posted 2023-02-17 任长江

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了JMS之ActiveMQ基本应用相关的知识，希望对你有一定的参考价值。

一、JMS：

概念：

JMS即Java Message Service（Java消息服务）应用程序接口，主要用于两个应用程序或分布式系统中的异步通信，JMS就像JDBC一样是一个Java平台的技术规范，JDBC就是用来访问不同关系数据库的API，而JMS就是用来进行消息收发的API。

体系架构：

1、JMS提供者：实现JMS规范的产品，下面讲的ActiveMQ就是其中一个，当然除了ActiveMQ外还有很多其他的提供者。

2、JMS客户：生产或消费消息的Java应用程序或对象。

3、JMS生产者：创建并发送消息的JMS客户。

4、JMS消费者：接收消息的JMS客户。

5、JMS消息：JMS客户之间传递的数据对象，消息类型可以为多种类型。

6、JMS队列：由于消息被生产者发送之后不会被消费者立即消费掉，所以需要先将消息放在JMS队列里等待消费者收取，消息被消费者消费之后就会从队列中移除。

7、JMS主题：支持发送消息给多个订阅者的机制，简单说就是生产者将消息发送到主题之后，主题可以将每个消息发给每个消费者。

对象模型：

1、连接工厂（ConnectionFactory）：用来创建JMS连接。

2、JMS连接（Connection）：JMS客户端与消息服务器之间的连接，即JMS客户端和消息服务器之间通信的高速公路。

3、JMS会话（Session）：即JMS客户端和消息服务器之间的一个线程，有了高速公路还需要有一辆卡车将货物运送过去啊。需要注意在实际的应用程序中会话可以支持事务，当选择支持事务的情况下，消息会先存放在会话中，当事务提交时会将消息一并发送，提交之前可以进行回滚操作。

4、JMS目的（Destination）：消息发布和接收的地点，可以是队列或主题，至于是队列还是主题要看消息通信的模型了（点对点模型与发布订阅模型），下边说明。

5、JMS生产者和消费者（MessageProducter，MessageConsumer）：消息发送和接收的客户端对象。

JMS消息通信模型：

1、点对点：

一个生产者向特定队列发送消息，一个消费者从该队列收取消息。该模式下生产者无需在消费者消费消息的情况下处于运行状态，消费者也无需在生产者发送消息的情况下处于运行状态。

2、发布订阅：

一个生产者向特定主题发送消息，0个或多个消费者从该主题获取消息，多个消费者之间的消费互不影响。该模式下发布和订阅者之间存在时间依赖性，订阅者需要先订阅对应主题的消息，否则将无法获取消息，也就是说你今天订阅人民日报的报纸，那么之前的报纸消息你是无法获取到的。

二、ActiveMQ

上边说了ActiveMQ是JMS规范的一个实现，那么也自然符合上边的相关体系和模型。下边通过代码看ActiveMQ的实现过程。

点对点模型

1、消息生产者：

/**
 * 消息生产者
 */
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; // 连接工厂
		Connection connection = null; // 连接
		Session session; // 会话 接受或者发送消息的线程
		Destination destination; // 消息的目的地
		MessageProducer messageProducer; // 消息生产者
		// 实例化连接工厂
		connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
		try 
			connection=connectionFactory.createConnection(); // 通过连接工厂获取连接
			connection.start(); // 启动连接
			session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建Session，参数为true表示开启事务
			destination=session.createQueue("Queue1"); // 创建消息队列
			messageProducer=session.createProducer(destination); // 创建消息生产者
			sendMessage(session, messageProducer); // 发送消息
			session.commit();
		 catch (Exception e) 
			e.printStackTrace();
		 finally
			if(connection!=null)
				try 
					connection.close();
				 catch (JMSException e) 
					e.printStackTrace();
				
			
		
	
	
	/**
	 * 发送消息
	 */
	public static void sendMessage(Session session,MessageProducer messageProducer)throws Exception
		for(int i=0;i<JMSProducer.SENDNUM;i++)
			TextMessage message=session.createTextMessage("ActiveMQ 发送的消息"+i);
			messageProducer.send(message); //消息生产者发送消息

2、消费者基本用法：

/**
 * 消息消费者
 */
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; // 消息的消费者
		
		// 实例化连接工厂
		connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
		try 
			connection=connectionFactory.createConnection();  // 通过连接工厂获取连接
			connection.start(); // 启动连接
			session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session，参数为false表示不开启事务
			destination=session.createQueue("Queue1");  // 创建连接的消息队列
			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();

3、消费者常用用法：

在上边的实现中消费者不断的从消息队列中收取消息，这样太耗费资源，通过我们会给消费者注册一个监听器，当监听到有新的消息时就通知消费者消费。

/**
 * 消息消费者
 */
public class JMSConsumer2 

	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; // 消息的消费者
		
		// 实例化连接工厂
		connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);	
		try 
			connection=connectionFactory.createConnection();  // 通过连接工厂获取连接
			connection.start(); // 启动连接
			session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
			destination=session.createQueue("Queue1");  // 创建连接的消息队列
			messageConsumer=session.createConsumer(destination); // 创建消息消费者
			messageConsumer.setMessageListener(new Listener()); // 注册消息监听
		 catch (JMSException e) 
			e.printStackTrace();

/**
 * 消息监听
 */
public class Listener implements MessageListener

	@Override
	public void onMessage(Message message) 
		try 
			System.out.println("收到的消息："+((TextMessage)message).getText());
		 catch (JMSException e) 
			e.printStackTrace();

订阅模式

1、消息发布者

/**
 * 消息生产者-消息发布者
 */
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; // 连接工厂
		Connection connection = null; // 连接
		Session session; // 会话 接受或者发送消息的线程
		Destination destination; // 消息的目的地
		MessageProducer messageProducer; // 消息生产者
		
		// 实例化连接工厂
		connectionFactory=new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD, JMSProducer.BROKEURL);
		try 
			connection=connectionFactory.createConnection(); //通过连接工厂获取连接
			connection.start(); //启动连接
			session=connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); //创建Session
			destination=session.createTopic("Topic1"); //这里创建的为消息主题
			messageProducer=session.createProducer(destination); //创建消息生产者
			sendMessage(session, messageProducer); //发送消息
			session.commit();
		 catch (Exception e) 
			e.printStackTrace();
		 finally
			if(connection!=null)
				try 
					connection.close();
				 catch (JMSException e) 
					e.printStackTrace();
				
			
		
	
	
	/**
	 * 发送消息
	 */
	public 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); //发送消息

2、消息订阅者一

/**
 * 消息消费者-消息订阅者一
 */
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; // 消息的消费者
		
		// 实例化连接工厂
		connectionFactory=new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD, JMSConsumer.BROKEURL);
		try 
			connection=connectionFactory.createConnection();  // 通过连接工厂获取连接
			connection.start(); // 启动连接
			session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
			destination=session.createTopic("Topic1"); //在这里创建的为消息主题
			messageConsumer=session.createConsumer(destination); // 创建消息消费者
			messageConsumer.setMessageListener(new Listener()); // 注册消息监听
		 catch (JMSException e) 
			e.printStackTrace();

/**
 * 消息监听-订阅者一
 */
public class Listener implements MessageListener

	@Override
	public void onMessage(Message message) 
		try 
			System.out.println("订阅者一收到的消息："+((TextMessage)message).getText());
		 catch (JMSException e) 
			e.printStackTrace();

3、消息订阅者二

/**
 * 消息消费者-消息订阅者二
 */
public class JMSConsumer2 

	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; // 消息的消费者
		
		// 实例化连接工厂
		connectionFactory=new ActiveMQConnectionFactory(JMSConsumer2.USERNAME, JMSConsumer2.PASSWORD, JMSConsumer2.BROKEURL);				
		try 
			connection=connectionFactory.createConnection();  // 通过连接工厂获取连接
			connection.start(); // 启动连接
			session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
			destination=session.createTopic("Topic1");
			messageConsumer=session.createConsumer(destination); // 创建消息消费者
			messageConsumer.setMessageListener(new Listener2()); // 注册消息监听
		 catch (JMSException e) 
			e.printStackTrace();

/**
 * 消息监听-订阅者二
 */
public class Listener2 implements MessageListener

	@Override
	public void onMessage(Message message) 
		try 
			System.out.println("订阅者二收到的消息："+((TextMessage)message).getText());
		 catch (JMSException e) 
			e.printStackTrace();

以上是关于JMS之ActiveMQ基本应用的主要内容，如果未能解决你的问题，请参考以下文章

深入浅出JMS--ActiveMQ简单介绍以及安装

MQ 之 ActiveMQ