ActiveMQ入门系列二：入门代码实例（点对点模式）
在上一篇《activemq入门系列一：认识并安装activemq（windows下）》中，大致介绍了activemq和一些概念，并下载、安装、启动他，还访问了他的控制台页面。
这篇，就用代码实例说下如何实现消息的生产和消费。
一、理论基础
同rabbitmq一样，activemq中也是有两种模式：
- 点对点模式（point to point，简写为ptp）
- 发布/订阅模式（publish & subscribe，简写为pub & sub）
通过我们知道了制造消息的应用叫生产者（producer），生产者在生产了消息后会发送消息到目的地（destination），到达消费和处理消息的应用（也就是消费者consumer）。这里的两种模式就通过对应不同的消息目的地（destination）来实现，ptp对应queue（队列）、pub&sub对应topic（主题）。
今天就详细介绍下ptp和queue，下一篇介绍pub & sub和topic。
在ptp模式的示意图：
- 消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息。
- 消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息。
- queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费、其它的则不能消费此消息了。
- 当消费者不存在时，消息会一直保存，直到有消费消费。
在ptp中，代码实现有两种方式：消费者主动消费和消费者监听消费，下面就分别说下。
二、消费者主动消费
主动消费是最基本也是最简单的消费方式，先上代码：
- 创建maven工程并引入依赖
<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()之后不会再消费其他消息了，即便后面再有消息被生产出来也不会再消费。也就是说只能在运行后消费一次消息，这个就是主动消费。
如果想要循环消费多次产生的消息的话，怎么办呢？请用下面的监听消费
三、消费者监听消费
还是先上代码，代码结构同主动消费类似，有细微差别，具体代码不贴了，可以到我的github或上获取源码
- 首先为了区分，我把队列名改了
public static final string queue_name = "ptp-listener-demo";//队列名
- 生产者和消费者的消息确认方式都改成了客户端手动确认，不再自动确认，手动确认有个好处就是可以防止消息没有被正常消费而丢失，这个同rabbitmq机制一样
session = connection.createsession(false, session.client_acknowledge);
- 生产者生产消息的时候，为了方便我改成了一次性发送10条
/** * 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的点对点模式就介绍完了。下一篇介绍发布订阅模式，敬请期待