特点
-
queue 是点对点模式,一条消息对应一个消费者,topic 是一对多模式,一条消息可能有一个或多个消费者
-
queue 模式消息再发送后消费者可以在之后的任意时间消费,topic 模式如果没有订阅者,消息就是废消息,会被丢弃。
-
queue 模式生产者与消费者之间没有时间相关性,topic 模式下生产者和消费者之间有一定的时间相关性,消费者只能接收到订阅之后的生产者发送的消息。
消息模型:
- 队列:Point-to-Point(P2P) --- 点对点(生产者发送一条消息到 queue,只有一个消费者能收到)
- 主题:Publish/Subscribe(Pub/Sub)--- 发布订阅(发布者发送到 topic 的消息,只有订阅了topic 的订阅者才会收到消息)
topic 消费的几种情况
- 0 个消费者,消息会被直接废止,
- n 个消费者,一部分消费,
- n 个消费者,全都消费了,
JMS 规范
JMS开发步骤;
-
创建连接工厂(ConnectionFactory)。
-
创建连接(Connection)。
-
创建会话(Session)。
-
创建 Topic(Destination)。
-
创建消息生产者(MessageProducer)。
-
创建消息消费者(MessageConsumer)。
-
发送消息。
-
接收消息。
-
关闭资源。
案例
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 测试消息队列
*
* @author Mr.css
* @version 2024-10-12 11:09
*/
public class ActiveMQSender {
private static final String USERNAME = "admin";
private static final String PASSWORD = "admin";
private static final String BROKER_URL = "tcp://localhost:61616";
private static final String TOPIC_NAME = "MyQueue";
public static void main(String[] args) throws Exception {
// 连接池
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
// 连接
Connection connection = factory.createConnection();
connection.start();
// 会话 - 自动确认 ACK
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic(TOPIC_NAME);
// 消息生产者 - 非持久化消息
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
TextMessage message = session.createTextMessage("Hello ActiveMQ!");
producer.send(message);
System.out.println("Sent message: " + message.getText());
session.close();
connection.close();
}
}
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 测试消息队列
*
* @author Mr.css
* @version 2024-10-12 11:09
*/
public class ActiveMQReceiver {
private static final String USERNAME = "admin";
private static final String PASSWORD = "admin";
private static final String BROKER_URL = "tcp://localhost:61616";
private static final String TOPIC_NAME = "MyQueue";
public static void main(String[] args) throws Exception {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
// 连接
Connection connection = factory.createConnection();
connection.start();
// 会话 - 自动确认
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic(TOPIC_NAME);
// 消息消费者
MessageConsumer consumer = session.createConsumer(destination);
while (true) {
// receive() 函数如果不带参数,则表示持续侦听消息
TextMessage message = (TextMessage) consumer.receive();
if (message != null) {
System.out.println("Received message: " + message.getText());
} else {
// 如果是持续接受消息,这里执行不到
break;
}
}
session.close();
connection.close();
}
}
消息侦听器的写法
前面的案例,可以让消息侦听,运行在指定的线程内部,编码的自由度更高写(案例中为主线程)。
我们一般不需要做到这种程度,侦听器的写法,可能更符合我们的编程习惯。
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 测试消息队列
*
* @author Mr.css
* @version 2024-10-12 11:09
*/
public class ActiveMQReceiver {
private static final String USERNAME = "admin";
private static final String PASSWORD = "admin";
private static final String BROKER_URL = "tcp://localhost:61616";
private static final String TOPIC_NAME = "MyQueue";
public static void main(String[] args) throws Exception {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
// 连接
Connection connection = factory.createConnection();
connection.start();
// 会话 - 自动确认
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createTopic(TOPIC_NAME);
// 消息消费者
MessageConsumer consumer = session.createConsumer(destination);
// 这里会启用子线程进行侦听,程序会继续向下执行
consumer.setMessageListener(message -> {
if(message instanceof TextMessage){
try {
System.out.println("Received message: " + ((TextMessage) message).getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
// 这里需要阻塞主线程,不然主线程跑完,消息侦听线程就跟着结束了,
// 想要关闭程序,控制台随便输入点什么,程序就会继续执行了,
// 以 springboot 项目为例,代码写到这就够了,后面的 close() 放到 容器销毁侦听事件中。
System.in.read();
consumer.close();
session.close();
connection.close();
}
}
标签:String,模式,topic,session,static,private,message,activemq,final
From: https://www.cnblogs.com/chenss15060100790/p/18524412