本节主要实现ActiveMQ-Queues点对点消息生产者及消费者的Receive+Listener两种接收方式
1,开启ActiveMQ服务-请参考 : ActiveMQ的下载,安装和启动 2,创建生产者,用于向指定队列发送测试数据 3,创建消费者-Receive方式,消费由生产者发送的队列的消息 4,创建消费者-Listener方式,消费由生产者发送的队列的消息 5,查看ActiveMQ后台,显示,生产者,消费者及消息的接收和处理状态
以下代码实现了:连接ActiveMQ,创建队列,发送消息等一些列操作(见代码注释)
package com.brave.ActiveMQ.Producer; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 消息生产者 * @author Brave * */ public class Producer { private static final String USERNAME = ActiveMQConnection.DEFAULT_USER; private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL; public static void main(String[] args) { ConnectionFactory connectionFactory; //连接工厂 Connection connection = null;// 连接 Session session; //会话 Destination destination; //消息地址 MessageProducer messageProducer; //消息生产者 // 实例化连接工厂 connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL); try { // 创建连接 connection = connectionFactory.createConnection(); // 连接开启 connection.start(); // 创建Session会话 // 参数1:是否开启事务 // 参数2:会话Session // Session.AUTO_ACKNOWLEDGE - 自动确认 消费者从receive或监听成功返回时,自动确认客户端收到消息 // Session.CLIENT_ACKNOWLEDGE - 客户通过acknowledge方法确认消息(会话层确认),确认一个即确认所有 // Session.DUPS_OK_ACKNOWLEDGE - 重复确认 session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 创建消息队列 destination = session.createQueue("TestQueue"); // 创建消息生产者 messageProducer = session.createProducer(destination); // 发送消息 sendMessage(session, messageProducer); // 由于加入了Session,需要提交 session.commit(); } catch (Exception e) { e.printStackTrace(); System.out.println("错误类名称 = " + e.getClass().getName()); System.out.println("错误原因 = " + e.getMessage()); }finally { //发送完毕后-关闭释放 if(connection != null){ try { connection.close(); connection = null; } catch (JMSException e) { e.printStackTrace(); } } } } /** * 发送消息 * @param session 会话 * @param messageProducer 消息生产者对象 * @throws JMSException */ private static void sendMessage(Session session, MessageProducer messageProducer) throws JMSException{ for(int i = 0; i < 10; i++){ TextMessage textMessage = session.createTextMessage("TestMessage:index = " + i); System.out.println("ActiveMQ-生产者-发送消息-index = " + i); messageProducer.send(textMessage); } } }运行生产者main方法 : 发10条消息,由于暂时没有消费者,消息将堆积在队列中
生产者发送消息Log:
生产者发送消息后台查询:
后台显示在TestQueue队列中有10条消息,并且未被消费者消费
以下代码实现一个消费者:以轮训读取方式获取指定队列来自生产者发送的消息(具体流程见代码注释)
package com.brave.ActiveMQ.Consumer_Receive; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; import javax.jms.Session; import javax.jms.TextMessage; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 消息消费者 * 方式一:Receive方式(不推荐) * * @author Brave * */ public class Consumer_Receive { private static final String USERNAME = ActiveMQConnection.DEFAULT_USER; private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL; public static void main(String[] args) { ConnectionFactory connectionFactory; // 连接工厂 Connection connection = null; // 连接 Session session; // 会话 Destination destination; // 消息地址 MessageConsumer messageConsumer; // 消息消费者 // 实例化连接工厂 connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL); try { // 创建连接 connection = connectionFactory.createConnection(); // 连接开启 connection.start(); // 创建Session会话 // 参数1:是否开启事务 // 参数2:会话Session // Session.AUTO_ACKNOWLEDGE - 自动确认 消费者从receive或监听成功返回时,自动确认客户端收到消息 // Session.CLIENT_ACKNOWLEDGE - 客户通过acknowledge方法确认消息(会话层确认),确认一个即确认所有 // Session.DUPS_OK_ACKNOWLEDGE - 重复确认 session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建连接消息队列 destination = session.createQueue("TestQueue"); // 创建消息消费者 messageConsumer = session.createConsumer(destination); // 接收数据-Receive方式-由于一直要轮训接收,不推荐 while(true){ TextMessage textMessage= (TextMessage)messageConsumer.receive(10000);//每10000毫秒接收一次消息 if(textMessage != null){ System.out.println("消费者-Receive方式-收到消息: " + textMessage.getText()); } } } catch (Exception e) { e.printStackTrace(); System.out.println("错误类名称 = " + e.getClass().getName()); System.out.println("错误原因 = " + e.getMessage()); }finally { //关闭释放 if(connection != null){ try { connection.close(); connection = null; } catch (JMSException e) { e.printStackTrace(); } } } } }消费者接收消息Log:
消费者接收消息后台查询:
后台显示,刚刚生产者发送的10调消息,已经被消费者消费了
以下代码实现一个消费者:以监听方式获取指定队列来自生产者发送的消息(具体流程见代码注释)
package com.brave.ActiveMQ.Consumer_Listener; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.MessageConsumer; import javax.jms.Session; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; /** * 消息消费者 * 方式二:Listener方式(推荐) * * @author Brave * */ public class Consumer_Listener { private static final String USERNAME = ActiveMQConnection.DEFAULT_USER; private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD; private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL; public static void main(String[] args) { ConnectionFactory connectionFactory; // 连接工厂 Connection connection = null; // 连接 Session session; // 会话 Destination destination; // 消息地址 MessageConsumer messageConsumer; // 消息消费者 // 实例化连接工厂 connectionFactory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKEURL); try { // 创建连接 connection = connectionFactory.createConnection(); // 连接开启 connection.start(); // 创建Session会话 // 参数1:是否开启事务 // 参数2:会话Session // Session.AUTO_ACKNOWLEDGE - 自动确认 消费者从receive或监听成功返回时,自动确认客户端收到消息 // Session.CLIENT_ACKNOWLEDGE - 客户通过acknowledge方法确认消息(会话层确认),确认一个即确认所有 // Session.DUPS_OK_ACKNOWLEDGE - 重复确认 session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建连接消息队列 destination = session.createQueue("TestQueue"); // 创建消息消费者 messageConsumer = session.createConsumer(destination); //Listener方式:注册消息监听器 messageConsumer.setMessageListener(new MessageListener()); } catch (Exception e) { e.printStackTrace(); System.out.println("错误类名称 = " + e.getClass().getName()); System.out.println("错误原因 = " + e.getMessage()); }finally { //关闭释放 // if(connection != null){ // try { // connection.close(); // connection = null; // } catch (JMSException e) { // e.printStackTrace(); // } // } } } } package com.brave.ActiveMQ.Consumer_Listener; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.TextMessage; /** * 消息监听器 * @author Brave * */ public class MessageListener implements javax.jms.MessageListener { @Override public void onMessage(Message message) { try { System.out.println("消费者-监听方式-收到消息: " + ((TextMessage)message).getText()); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }我们再次使用生产者代码生产10条测试数据
启动监听方式的消费者,并查看Log:
消费者接收消息后台查询:
后台显示:新创建的10条数据已成功被消费了
下载
GitHub下载
更新记录:
20170503:
添加项目结构和项目依赖说明配图 重构项目,更新代码段 更新项目地址为重构项目 更新GitHub项目地址为重构项目