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(); } }
相关推荐
方新德 2020-04-20
胡献根 2020-02-23
胡献根 2020-02-13
WindyQCF 2014-05-12
Jolestar 2020-02-13
胡献根 2020-07-18
胡献根 2020-07-05
jiangtie 2020-06-10
onlylixiaobei 2020-06-09
xinglun 2020-06-02
方新德 2020-05-31
Java高知 2020-05-20
Java高知 2020-05-08
Java高知 2020-05-03
onlylixiaobei 2020-05-02
Java高知 2020-04-22
胡献根 2020-04-22
heweiyabeijing 2020-04-21