JMS ActiveMQ Queue(点对点模型)实例(二)

一.使用消息队列模型发送消息至activeMQ(生产者开发)

package com.gosun.activemq;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;

/**
 * 消息队列模型发送消息至activeMQ
 * @author Ickes
 */
public class QueueSend {
	public static void main(String[] args) throws Exception {
		//第一步:根据url创建一个jms Connection。 
		ActiveMQConnectionFactory connectionfactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
		Connection connection = connectionfactory.createConnection(); 
		connection.start();  
		//第二步:根据connection获取session
		Session session =connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
		//第三步:消息的目的地
		Destination destination = new ActiveMQQueue("gosun"); 
		//第四步:创建消息生产者
		MessageProducer producer = session.createProducer(destination);  
		//设置不持久化
		producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		//第五步:创建消息
		Message msg = session.createTextMessage("JMS 告诉你我是ICKES");
		//第六步:生产者向JMS发送消息到队列
		producer.send(msg);
        //第七步:关闭连接
		session.close();  
        connection.close(); 
	}
}

 第二步:Session session =connection.createSession(false,Session.AUTO_ACKNOWLEDGE);解释如下

在connection的基础上创建一个session,同时设置是否支持事务ACKNOWLEDGE标识。 

  • AUTO_ACKNOWLEDGE:自动确认模式。一旦接收方应用程序的方法调用从处理消息处返回,会话对象就会确认消息的接收。  
  • CLIENT_ACKNOWLEDGE:客户端确认模式。会话对象依赖于应用程序对被接收的消息调用一个acknowledge()方法。一旦这个方法被调用,会话会确认最后一次确认之后所有接收到的消息。这种模式允许应用程序以一个调用来接收,处理并确认一批消息。注意:在管理控制台中,如果连接工厂的Acknowledge Policy(确认方针)属性被设置为"Previous"(提前),但是你希望为一个给定的会话确认所有接收到的消息,那么就用最后一条消息来调用acknowledge()方法。  
  • DUPS_OK_ACKNOWLEDGE:允许副本的确认模式。一旦接收方应用程序的方法调用从处理消息处返回,会话对象就会确认消息的接收;而且允许重复确认。在需要考虑资源使用时,这种模式非常有效。注意:如果你的应用程序无法处理重复的消息的话,你应该避免使用这种模式。如果发送消息的初始化尝试失败,那么重复的消息可以被重新发送。 
  • SESSION_TRANSACTED

 

二、消息消费者,手动接收示例(消费者)

package com.gosun.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消息消费者,手动接收示例
 * @author Ickes
 */
public class QueuesAccept {
	public static void main(String[] args) throws Exception {
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
		Connection conn = connectionFactory.createConnection();  
		Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); 
		conn.start();
		//消息目的地
		Destination dest = session.createQueue("gosun");
		//消息消费者
		MessageConsumer consumer = session.createConsumer(dest);
		//接收消息,超时时间为10秒,先手动接受JMS消息,这儿可以用监听 
		TextMessage textMessage = (TextMessage) consumer.receive(10*1000);
		String text = textMessage.getText();  
	    System.out.println("接收到的消息为:" + text);  
	    //关闭通道
	    consumer.close();  
	    session.close();  
	    conn.close();  
	}
}

 

三、消费者监听器模式处理消息

package com.gosun.activemq;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 使用监听器,自动接收消息
 * @author Lenovo
 *
 */
public class QueuesAcceptListener implements MessageListener{
	
	@Override
	public void onMessage(Message message) {
		 TextMessage text = (TextMessage) message;
		 try {
			System.out.println(text.getText());
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 测试代码
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
		Connection conn = connectionFactory.createConnection();  
		Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); 
		conn.start();
		//消息目的地
		Destination dest = session.createQueue("gosun");
		//消息消费者
		MessageConsumer consumer = session.createConsumer(dest);
		consumer.setMessageListener(new QueuesAcceptListener());
		//这里不能关闭连接,一旦关闭监听器也就关闭,那就接收不到消息了
	}

}

 

四、生产者发送消息到主题中

package com.gosun.activemq.topic;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQTopic;

/**
 * 生产者发送消息到主题中
 * @author Ickes
 *
 */
public class TopicSend {
	public static void main(String[] args) throws Exception {
		// 第一步:根据url创建一个jms Connection。
		ActiveMQConnectionFactory connectionfactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
		Connection connection = connectionfactory.createConnection();
		connection.start();
		// 第二步:根据connection获取session
		Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
		// 第三步:创建一个Topic
		Topic topic= new ActiveMQTopic("testTopic");  
		// 第四步:创建生产者用于将消息发送至主题
		MessageProducer producer = session.createProducer(topic); 
		// 设置不持久化
		producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		// 第五步:创建消息
		Message msg = session.createTextMessage("JMS 告诉你我是ICKES");
		producer.send(msg);
		//第七步:关闭连接
		session.close();
		connection.close();
	}
}

五、消费者从主题中订阅消息

package com.gosun.activemq.topic;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * 消费者从主题中订阅消息
 * @author Ickes
 *
 */
public class TopicAccept {
	public static void main(String[] args) throws Exception {
		ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
		Connection conn = connectionFactory.createConnection();  
		Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); 
		conn.start();
		//主题目的地
		Topic topic =session.createTopic("testTopic");
		//注册订阅者
		MessageConsumer consumer = session.createConsumer(topic);
		//手动获取消息
		TextMessage textMessage = (TextMessage) consumer.receive(10*1000);
		String text = textMessage.getText();  
	    System.out.println("接收到的消息为:" + text);  
	    //关闭通道
	    consumer.close();  
	    session.close();  
	    conn.close();  
		
		
	}
}

相关推荐