JMS学习笔记(二)——Queue和Topic
2014-04-11 13:24
330 查看
一,点到点和发布/订阅详解
在点对点或队列模型下,一个生产者向一个特定的队列(queue)发布消息,一个消费者从该队列中读取消息。这里,生产者知道消费者的队列,并直接将消息发送到消费者的队列。可概括为:1,每条消息只有一个消费者,如果一条消息被消息者接收,那么其他的消费者就不能得到这条消息了。
2,生产者不需要在接收者消费该消息期间处于运行状态,接收者也同样不需要在消息发送时处于运行状态。队列是先进先出的,消费者可以接收到它离线时的消息。
3,每一个成功处理的消息都由接收者签收,并返回一个标识,表明哪个消息已经被接收
发布者/订阅者模型支持向一个特定的消息主题发布消息。0或多个订阅者可能对接收来自特定消息主题的消息感兴趣。在这种模型下,发布者和订阅者彼此不知道对方。这种模式好比是匿名公告板。可概括为:
1,多个消费者可以获得消息
2,在发布者和订阅者之间存在时间依赖性。发布者需要建立一个订阅(subscription),以便客户能够订阅。订阅者必须保持持续的活动状态以接收消息,除非订阅者建立了持久的订阅,在这种情况下,在订阅者未连接时发布的消息将在订阅者重新连接时重新发布。
二,Demo程序
下载apache-activemq-5.9.0,解压,可以在其中找到activemq-all-5.9.0.jar。建立一个java项目,通过bulidpath将activemq-all-5.9.0.jar引用到项目中。运行apache-activemq-5.9.0目录下的bin/activemq.bat,在浏览器中访问http://localhost:8161/admin/,帐号密码均为admin,如果访问成功,说明activemq服务启动成功。下面编写程序1,Queue模式
消息发送者:
package com.rick.queue; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; public class Sender { private static final int SEND_NUMBER = 5; public static void main(String[] args) { System.out.println("QueueSender启动了..."); // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory; // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = null; // Session: 一个发送或接收消息的线程 Session session; // Destination :消息的目的地;消息发送给谁. Destination destination; // MessageProducer:消息发送者 MessageProducer producer; // TextMessage message; // 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616"); try { // 构造从工厂得到连接对象 connection = connectionFactory.createConnection(); // 启动 connection.start(); // 获取操作连接 session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE); // 获取session注意参数值FirstQueue是一个服务器的queue,须在在ActiveMq的console配置 destination = session.createQueue("FirstQueue"); // 得到消息生成者【发送者】 producer = session.createProducer(destination); // 设置不持久化,此处学习,实际根据项目决定 producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 构造消息,此处写死,项目就是参数,或者方法获取 sendMessage(session, producer); session.commit(); } catch (Exception e) { e.printStackTrace(); } } public static void sendMessage(Session session, MessageProducer producer) throws Exception { for (int i = 1; i <= SEND_NUMBER; i++) { TextMessage message = session .createTextMessage("ActiveMq 发送的消息" + i); // 发送消息到目的地方 System.out.println("发送消息:" + "ActiveMq 发送的消息" + i); producer.send(message); } } }消息接收者:
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.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; public class Receiver { public static void main(String[] args) { System.out.println("Receiver启动了..."); // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory; // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = null; // Session: 一个发送或接收消息的线程 Session session; // Destination :消息的目的地;消息发送给谁. Destination destination; // 消费者,消息接收者 MessageConsumer consumer; connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616"); try { // 构造从工厂得到连接对象 connection = connectionFactory.createConnection(); // 启动 connection.start(); // 获取操作连接 session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 获取session注意参数值FirstQueue是一个服务器的queue,须在在ActiveMq的console配置 destination = session.createQueue("FirstQueue"); consumer = session.createConsumer(destination); TextMessage message; while (true) { message = (TextMessage) consumer.receive(); System.out.println("收到消息" + message.getText()); } } catch (Exception e) { e.printStackTrace(); } finally { try { if (null != connection) connection.close(); } catch (Throwable ignore) { } } } }先运行Sender,再运行Receiver,也可先运行Receiver。如果要在两台电脑上测试,可将Receiver放到另外一台电脑上运行,并且要把ip更改为activemq所在服务器的ip。
运行结果:
2,Pub/Sub,发布/订阅模式
消息发送者:
package com.rick.topic; import java.io.Serializable; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.MessageProducer; import javax.jms.ObjectMessage; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; public class TopicSender { private static final int SEND_NUMBER = 5; public static void main(String[] args) { System.out.println("TopicSender启动了..."); // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory; // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = null; // Session: 一个发送或接收消息的线程 Session session; // Destination :消息的目的地;消息发送给谁. Destination destination; // MessageProducer:消息发送者 MessageProducer producer; // TextMessage message; // 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616"); try { // 构造从工厂得到连接对象 connection = connectionFactory.createConnection(); // 启动 connection.start(); // 获取操作连接 session = connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE); // 获取session注意参数值mytopic是一个服务器的topic,须在在ActiveMq的console配置 destination = session.createTopic("mytopic"); // 得到消息生成者【发送者】 producer = session.createProducer(destination); // 设置不持久化,此处学习,实际根据项目决定 producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 构造消息,此处写死,项目就是参数,或者方法获取 sendMessage(session, producer); session.commit(); } catch (Exception e) { e.printStackTrace(); } finally { try { if (null != connection) connection.close(); } catch (Throwable ignore) { } } } public static void sendMessage(Session session, MessageProducer producer) throws Exception { for (int i = 1; i <= SEND_NUMBER; i++) { TextMessage message = session.createTextMessage("topic:" + i); System.out.println("Sender发送消息:" + "topic:" + i); producer.send(message); } } }消息接收者,可创建多个:
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 javax.jms.Topic; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; public class TopicReceiver { public static void main(String[] args) { // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory; // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = null; // Session: 一个发送或接收消息的线程 Session session; // Destination :消息的目的地;消息发送给谁. Destination destination; // 消费者,消息接收者 MessageConsumer consumer; connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616"); try { // 构造从工厂得到连接对象 connection = connectionFactory.createConnection(); // 启动 connection.start(); // 获取操作连接 session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 获取session注意参数值mytopic是一个服务器的topic,须在在ActiveMq的console配置 destination = session.createTopic("mytopic"); consumer = session.createConsumer(destination); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { if(message instanceof TextMessage){ try { TextMessage text=(TextMessage)message; System.out.println("收到消息" + text.getText()); } catch (JMSException e) { e.printStackTrace(); } } } }); } catch (Exception e) { e.printStackTrace(); } } }
这种方式下,必须先运行Receiver,再运行发送者,否则会收不到消息。但是如果配置了持久化的订阅者,就可以先运行发送者,这点后面会再说明。
运行结果:
可以访问http://localhost:8161/admin/查看你创建的Queue和Topic,并且也可在页面上向Queue和Topic中发送消息
注:1,上述两个例子中所写的Receiver,一个是用while循环配合consumer.receiver()方法是实现,另一个采用consumer.setMessageListener()的方法实现,两种方法均可,但如果两者都用的话就会出错,如下面的代码就会出错:
consumer = session.createConsumer(queue); consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { try { message = (TextMessage) consumer.receive(); System.out.println("收到消息:"+ ((TextMessage) message).getText()); } catch (JMSException e) { e.printStackTrace(); } } });
异常信息:
总之,使用了setMessageListener()形参message就已经有值了,就不能再使用consumer.receiver()了。
2,上面的程序中都没有调用session.close()和connection.close(),所以程序一直处于运行状态(jms内部实现),如果想停止jms程序,就可通过调用session.close()和connection.close()方法来实现,具体根据需求而定。
相关文章推荐
- STL学习笔记12— —特殊容器queue
- JMS中queue和topic区别
- Jms两种message传输方式Topic和Queue的比较
- nginx 源码学习笔记(九)——基本容器——queue
- JMS消息类型模型[queue和topic的区别]
- Java并发学习笔记(八)-LinkedBlockingQueue
- STL学习笔记— —特殊容器priority_queue
- JMS(java消息服务)学习笔记
- 小BO学习笔记之ConcurrentLinkedQueue和LinkedBlockingQueue的使用比较
- JMS学习的个人理解笔记
- GCD学习笔记(二):dispatch_queue_create
- ArrayBlockingQueue学习笔记
- JMS学习6 - 一个topic上多个consumer
- ListBuffer、ArrayBuffer、Queue、Stack操作代码实战之Scala学习笔记-30
- nginx 源码学习笔记(九)——基本容器——queue
- 【Java数据结构学习笔记之三】Java数据结构与算法之队列(Queue)实现
- GCD学习笔记(三):Main Dispatch Queue/Global Dispatch Queue
- JMS和ActiveMQ学习笔记一
- Jms两种message传输方式Topic和Queue的比较
- c++学习笔记(2)——有关queue