ActiveMQ学习笔记(二)
程序员文章站
2022-07-15 09:06:47
...
使用主题Topic发布/订阅消息
多个消费着订阅同一主题,生产者发布该主题消息,此时多个消费者会同时收到消息,即一个生产者对应多个消费者。如果没有消费者订阅主题(即程序没有启动或没有在监听),生产者发送的消费是废消息,没有任何人能够收到。即便消费者后来订阅了也不会收到历史消息(持久订阅除外)。
消费者订阅主题
这里要先启动消费者,只有当消费者订阅了主题才能收消息
package cn.fg.jsm.listener;
import java.io.IOException;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
public class TopicListener {
public static void main(String[] args) throws JMSException, IOException {
//1.创建连接工厂,brokerURL(可以理解为实例):ActiveMQ实例的地址
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.1.3:61616");
//2.通过工厂创建连接
Connection connection = factory.createConnection();
connection.start(); //开始连接
//3.创建回话,两个参数:transacted 事务,acknowledgeMode 签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地(队列或主题)
Topic topic = session.createTopic("topic01"); //创建主题
//5.创建消费者,参数:destination 目的地
MessageConsumer messageConsumer = session.createConsumer(topic);
//6.创建监听器,MessageListener是一个接口,可以自己自定义监听类实现接口,这里就使用一个内部类演示
messageConsumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
//消息到了之后的逻辑处理
if (message != null && message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println(textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
//7.释放资源
System.in.read(); //控制台按任意键继续,这里阻塞一下,如果关闭了资源,消息就收不到了
messageConsumer.close();
session.close();
connection.close();
}
}
生产者发布主题消息
package cn.fg.jsm.producer;
import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
public class ProducerTopic {
public static void main(String[] args) throws JMSException {
//1.创建连接工厂
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://192.168.1.3:61616");
//2.通过工厂创建连接
Connection connection = factory.createConnection();
connection.start(); //开始连接
//3.创建回话,两个参数:transacted 事务,acknowledgeMode 签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4.创建目的地(队列或主题)
Topic topic = session.createTopic("topic01"); //创建主题
//5.创建消息生产者,参数:destination 目的地
MessageProducer messageProducer = session.createProducer(topic);
//6.创建消息
TextMessage textMessage = session.createTextMessage("这是一个消息" + System.currentTimeMillis());
//7.发送消息
messageProducer.send(textMessage);
//8.释放资源
messageProducer.close();
session.close();
connection.close();
System.out.println("消息发送完毕");
}
}