在上一篇《ActiveMQ入门系列一:认识并安装ActiveMQ(Windows下)》中,大体介绍了ActiveMQ和一些概念,并下载、安装、启动他,还访问了他的控制台页面。html
这篇,就用代码实例说下如何实现消息的生产和消费。java
1、理论基础git
同RabbitMQ同样,ActiveMQ中也是有两种模式:github
经过上一篇咱们知道了制造消息的应用叫生产者(Producer),生产者在生产了消息后会发送消息到目的地(Destination),到达消费和处理消息的应用(也就是消费者Consumer)。这里的两种模式就经过对应不一样的消息目的地(Destination)来实现,PTP对应Queue(队列)、Pub&Sub对应Topic(主题)。apache
今天就详细介绍下PTP和Queue,下一篇介绍Pub & Sub和Topic。session
在PTP模式的示意图:maven
在PTP中,代码实现有两种方式:消费者主动消费和消费者监听消费,下面就分别说下。tcp
2、消费者主动消费ide
主动消费是最基本也是最简单的消费方式,先上代码:测试
<dependency> <groupId>org.apache.activemq</groupId> <artifactId>activemq-core</artifactId> <version>5.7.0</version> </dependency>
package com.sam.ptp;
import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; /** * @author JAVA开发老菜鸟 * */ public class Producer { public static final String QUEUE_NAME = "ptp-demo";//队列名 public void producer(String message) throws JMSException { ConnectionFactory factory = null; Connection connection = null; Session session = null; MessageProducer producer = null; try { /** * 1.建立链接工厂 * 建立工厂,构造方法有三个参数:分别是用户名、密码、链接地址 * 无参构造:有默认的链接地址,localhost * 一个参数:无验证模式,无用户的认证 * 三个参数:有认证和链接地址,我这里使用三个参数的构造方法 */ factory = new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616"); /** * 2.建立链接,有两个方法(我这里使用无参数的) * 无参数 * 有参数:用户名、密码; */ connection = factory.createConnection(); /** * 3.启动链接 * 生产者能够不用调用start()方法启动,由于在发送消息的时候回进行检查 * 若是未启动链接,会自动启动。 * 若是有特殊配置,须要配置完成后再启动链接 */ connection.start(); /** * 4.用链接建立会话 * 有两个参数:是否须要事务、消息确认机制 * 若是支持事务,对于生产者来讲第二个参数就无效了,这个时候第二个参数建议传入Session.SESSION_TRANSACTED * 若是不支持事务,第二个参数有效且必须传递 * * AUTO_ACKNOWLEDGE:自动确认,消息处理后自动确认(商业开发不推荐) * CLIENT_ACKNOWLEDGE:客户端手动确认,消费者处理后必须手动确认 * DUPS_OK_ACKNOWLEDGE:有副本的客户端手动确认,消息能够屡次处理(不建议) */ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); /** * 5.用会话建立目的地(队列)、生产者、消息 * 队列名是队列的惟一标记 * 建立生产者的时候能够指定目的地,也能够在发送消息的时候再指定 */ Destination destination = session.createQueue(QUEUE_NAME); producer = session.createProducer(destination); TextMessage textMessage = session.createTextMessage(message); /** * 6.生产者发送消息到目的地 */ producer.send(textMessage); System.out.println("消息发送成功"); } catch(Exception ex){ throw ex; } finally { /** * 7.释放资源 */ if(producer != null){ producer.close(); } if(session != null){ session.close(); } if(connection != null){ connection.close(); } } } public static void main(String[] args){ Producer producer = new Producer(); try{ producer.producer("hello, activemq"); } catch (Exception ex){ ex.printStackTrace(); } } }
package com.sam.ptp; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; /** * @author JAVA开发老菜鸟 * * 主动消费 */ public class Consumer { public String consumer() throws JMSException { ConnectionFactory factory = null; Connection connection = null; Session session = null; MessageConsumer consumer = null; try { factory = new ActiveMQConnectionFactory("admin","admin","tcp://localhost:61616"); connection = factory.createConnection(); /** * 消费者必须启动链接,不然没法消费 */ connection.start(); session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); Destination destination = session.createQueue(Producer.QUEUE_NAME); consumer = session.createConsumer(destination); /** * 获取队列消息 */ Message message = consumer.receive(); String text = ((TextMessage) message).getText(); return text; } catch(Exception ex){ throw ex; } finally { /** * 7.释放资源 */ if(consumer != null){ consumer.close(); } if(session != null){ session.close(); } if(connection != null){ connection.close(); } } } public static void main(String[] args){ Consumer consumer = new Consumer(); try{ String message = consumer.consumer(); System.out.println("消息消费成功:" + message); } catch (Exception ex){ ex.printStackTrace(); } } }
好,这样代码就写好了,咱们来测试下。
1.先运行生产者,我发现报错了。。。
好吧,原来是我此次没有启动ActiveMQ,被本身蠢哭了。。。
启动ActiveMQ以后,再运行生产者,成功了。
去看下控制台页面的变化,队列里面多了个“ptp-demo”队列,这个就是咱们生产者代码里面的队列名,而且能看到该队列的基本状况:
从左到右依次为,有待消费的消息1条、消费者0个、已经发送的消息1条、已经消费的消息0条
2.接下来运行消费者,成功
再去看下控制台页面,发现队列信息变了,从左到右依次为:有待消费的消息0条、消费者0个、已经发送的消息1条、已经消费的消息1条
也就是说,消息真的被消费了!
代码写完了,也按照预期执行完了,咱们如今再回过头来分析下消费者的代码,会发现他在consumer.receive()以后不会再消费其余消息了,即使后面再有消息被生产出来也不会再消费。也就是说只能在运行后消费一次消息,这个就是主动消费。
若是想要循环消费屡次产生的消息的话,怎么办呢?请用下面的监听消费
3、消费者监听消费
仍是先上代码,代码结构同主动消费相似,有细微差异,具体代码不贴了,能够到个人GitHub或码云上获取源码
public static final String QUEUE_NAME = "ptp-listener-demo";//队列名
session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
/** * 6.建立消息而且生产者发送消息到目的地 */ for(int num = 0; num < 10; num++){ TextMessage textMessage = session.createTextMessage(message + num); producer.send(textMessage); System.out.println("消息发送成功"+textMessage.getText()); }
/** * 注册监听器,队列中的消息变化会自动触发监听器,接收并自动处理消息 * * 监听器一旦注册,永久有效,一直到程序关闭 * 监听器能够注册多个,至关于集群 * activemq自动轮询多个监听器,实现并行处理 */ consumer.setMessageListener(new MessageListener() { @Override public void onMessage(Message message) { try { //须要手动确认消息 message.acknowledge(); TextMessage om = (TextMessage) message; String data = om.getText(); System.out.println(data); } catch (JMSException e) { e.printStackTrace(); } } });
执行生产者:
执行消费者,消息所有被消费了:
再执行2遍生产者,消息一样都被消费了。
控制台页面多了个队列,因为监听中的消费者没有关闭,所以这里能看到消费者数量为1,我执行了三遍生产者,所以消息有30条。
还没完,继续...
咱们此次先启动2个消费者,而后启动生产者
两个生产者分别消费了消息0,2,4,6,8和1,3,5,7,9
也就是说两个消费者都监听到了消息,而且activemq自动轮询两个监听器发送消息。
好,到这里,ActiveMQ的点对点模式就介绍完了。下一篇介绍发布订阅模式,敬请期待