JMS之ActiveMQ基本应用

Posted 任长江

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基本应用的主要内容,如果未能解决你的问题,请参考以下文章

消息中间之ActiveMQ

activemq复习1

JMS消息服务之:ActiveMQ

activemq jms使用

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

MQ 之 ActiveMQ