JMS之ActiveMQ基本应用

2019-11-14 11:29:07

字体：大中小

来源：转载

供稿：网友

一、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();		}	}}