本文共 3296 字,大约阅读时间需要 10 分钟。
发布/订阅模式传递域的特点如下:
JMS规范允许客户创建持久订阅,这在一定程度上放松了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。类似微信公众号,关注之后可以查看发布过的所有消息。
发布主题生产者:
package pres.zhang.activemq;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class JmsProduce_topic { public static final String ACTIVEMQ_URL = "tcp://localhost:61616"; public static final String TOPIC_NAME = "topic_acton"; public static void main(String[] args) throws JMSException { //1.创建连接工厂,按照给定的url地址,采用默认用户名和密码 ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL); //2.通过连接工厂,获得连接connection,并启动访问 Connection connection = activeMQConnectionFactory.createConnection(); connection.start(); //3.创建回话session //两个参数 1.事务 2.签收 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //4.创建目的地(具体是队列还是主题topic) Topic topic = session.createTopic(TOPIC_NAME); //创建消息的生产者 MessageProducer messageProducer = session.createProducer(topic); //通过使用messageProducer生产3条消息发送到MQ的队列里面 for (int i = 0; i < 3; i++) { //7.创建消息 TextMessage textMessage = session.createTextMessage("TOPIC_NAME---" + i);//可以理解为字符串 //8.通过messageProducer发送给MQ messageProducer.send(textMessage); } //9.关闭资源 messageProducer.close(); session.close(); connection.close();; System.out.println("消息发布到TOPIC完成!"); }}
主题消费者
package pres.zhang.activemq;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;import java.io.IOException;public class JmsConsumer_topic { public static final String ACTIVEMQ_URL = "tcp://localhost:61616"; public static final String TOPIC_NAME = "topic_acton"; public static void main(String[] args) throws JMSException, IOException { //1.创建连接工厂,按照给定的url地址,采用默认用户名和密码 ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL); //2.通过连接工厂,获得连接connection,并启动访问 Connection connection = activeMQConnectionFactory.createConnection(); connection.start(); //3.创建回话session //两个参数 1.事务 2.签收 Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); //4.创建目的地(具体是队列还是主题topic) Topic topic = session.createTopic(TOPIC_NAME); //5.创建消费者 MessageConsumer messageConsumer = session.createConsumer(topic); //6.监听消息 messageConsumer.setMessageListener(message -> { if(message != null && message instanceof TextMessage){ TextMessage textMessage = (TextMessage) message; try { System.out.println("监听器接收消息:" + textMessage.getText()); } catch (JMSException e) { e.printStackTrace(); } } }); //保证控制台运行 System.in.read(); messageConsumer.close(); session.close(); connection.close(); }}
注意:先启动订阅,再启动消费者:
监听器接收消息:TOPIC_NAME---0监听器接收消息:TOPIC_NAME---1监听器接收消息:TOPIC_NAME---2
转载地址:http://ufpqb.baihongyu.com/