本节主要ActiveMQ-Topic模式实现消息的订阅和发布 本节代码基于2,ActiveMQ-Queues点对点消息-Receive+Listener方式 修改而来
1,消息发布者:
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"); // 创建消息发布 destination = session.createTopic("TestTopic"); // 创建消息生产者 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); } } }2,消息订阅者
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; /** * 消息订阅者1 * 方式二: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"); // 创建消息订阅 destination = session.createTopic("TestTopic"); // 创建消息消费者 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; /** * 消息订阅者1 * @author Brave * */ public class MessageListener implements javax.jms.MessageListener { @Override public void onMessage(Message message) { try { System.out.println("消息订阅者1-监听方式-收到消息: " + ((TextMessage)message).getText()); } catch (JMSException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }1,启动两个消息订阅者服务
INFO | Successfully connected to tcp://localhost:61616此时进入后台,选择Topic标签查看订阅发布视图
此时,后台显示有一个主题叫做TestTopic,同时又两个消费者在线,即消息订阅者
2,启动消息发布者服务,发布消息
INFO | Successfully connected to tcp://localhost:61616 ActiveMQ-生产者-发布消息-index = 0 ActiveMQ-生产者-发布消息-index = 1 ActiveMQ-生产者-发布消息-index = 2 ActiveMQ-生产者-发布消息-index = 3 ActiveMQ-生产者-发布消息-index = 4 ActiveMQ-生产者-发布消息-index = 5 ActiveMQ-生产者-发布消息-index = 6 ActiveMQ-生产者-发布消息-index = 7 ActiveMQ-生产者-发布消息-index = 8 ActiveMQ-生产者-发布消息-index = 9此时进入后台,选择Topic标签查看订阅发布视图
此时,后台显示主题TestTopic,有一个生产者,生产了10条消息,被消费了20次,即两个订阅者各10次
查看订阅者服务Log:
INFO | Successfully connected to tcp://localhost:61616 消息订阅者1-监听方式-收到消息: TestMessage:index = 0 消息订阅者1-监听方式-收到消息: TestMessage:index = 1 消息订阅者1-监听方式-收到消息: TestMessage:index = 2 消息订阅者1-监听方式-收到消息: TestMessage:index = 3 消息订阅者1-监听方式-收到消息: TestMessage:index = 4 消息订阅者1-监听方式-收到消息: TestMessage:index = 5 消息订阅者1-监听方式-收到消息: TestMessage:index = 6 消息订阅者1-监听方式-收到消息: TestMessage:index = 7 消息订阅者1-监听方式-收到消息: TestMessage:index = 8 消息订阅者1-监听方式-收到消息: TestMessage:index = 9 INFO | Successfully connected to tcp://localhost:61616 消息订阅者2-监听方式-收到消息: TestMessage:index = 0 消息订阅者2-监听方式-收到消息: TestMessage:index = 1 消息订阅者2-监听方式-收到消息: TestMessage:index = 2 消息订阅者2-监听方式-收到消息: TestMessage:index = 3 消息订阅者2-监听方式-收到消息: TestMessage:index = 4 消息订阅者2-监听方式-收到消息: TestMessage:index = 5 消息订阅者2-监听方式-收到消息: TestMessage:index = 6 消息订阅者2-监听方式-收到消息: TestMessage:index = 7 消息订阅者2-监听方式-收到消息: TestMessage:index = 8 消息订阅者2-监听方式-收到消息: TestMessage:index = 9LOG显示,生产者发布了10条消息,两个消费者各消费了10条消息
至此,ActiveMQ-Topic消息订阅发布模式完成
下载
GitHub下载
