ActivityMQ入门案例

1.下载ActivityMQ

http://activemq.apache.org/download-archives.html

下载的时候需要注意你下载的版本所需要的JDK版本支持情况  jdk1.6 支持 ActivityMQ5.4  5.5

2.启动ActivityMQ

解压apache-activemq-5.4.3-bin.zip到指定目录双击apache-activemq-5.4.3\bin\activemq.bat 如果没有报错说明启动成功

启动ActiveMQ以后，登陆：http://localhost:8161/admin/，创建一个Queue，命名为FirstQueue，此队列在代码中会使用到

3.Myeclipse中创建项目并导入apache-activemq-5.4.3\activemq-all-5.4.3.jar  该包

4.编码

Receiver.java

package com.test.mq; 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.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; public class Receiver { 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); // 获取队列 注意参数值FirstQueue是一个服务器的queue，须在在ActiveMq的console配置 destination = session.createQueue("FirstQueue"); consumer = session.createConsumer(destination); while (true) { //设置接收者接收消息的时间，为了便于测试，这里谁定为100s TextMessage message = (TextMessage) consumer.receive(100000); if (null != message) { System.out.println("收到消息" + message.getText()); } else { break; } } } catch (Exception e) { e.printStackTrace(); } finally { try { if (null != connection) connection.close(); } catch (Throwable ignore) { } } } }

Sender.java

 

package com.test.mq; 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) { // 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://localhost: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(); } 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("ActiveMq 发送的消息" + i); // 发送消息到目的地方 System.out.println("发送消息：" + "ActiveMq 发送的消息" + i); producer.send(message); } } }

 注:首先启动Receiver.java 监听在指定端口等待消息   然后启动Sender.java发送消息