ActiveMQ实战之 Queue点对点消息

时间:2021-06-16 20:34:39
前言:ActiveMQ消息模式点对点编码

运行:先运行消费者在开启消息生产者即可接收到消息

消息生产者

/**
* @摘要 测试发送单条数据的类
*/
public class ZMQOneSendTest{
public static void main(String[] args) throws Exception {
//1 .开始建立连接时间
Long nStartTime = System.currentTimeMillis(); // 2.建立连接工厂
org.apache.activemq.ActiveMQConnectionFactory nFactory = new org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616"); // 3.用工厂建立Connection连接
javax.jms.Connection nConnection = nFactory.createConnection(); // 4.创建会话Session,不启动事务,签收模式为自动确认模式
javax.jms.Session nSession = nConnection.createSession(false,javax.jms.Session.AUTO_ACKNOWLEDGE); // 5.创建队列
javax.jms.Destination nDestination = nSession.createQueue("123"); // 6.创建生产者
javax.jms.MessageProducer nProducer = nSession.createProducer(nDestination); // 7.设置持久化
nProducer.setDeliveryMode(javax.jms.DeliveryMode.PERSISTENT); // 8.启动连接
nConnection.start(); // 9.查询开始时间
Long nStartSelectTime = System.currentTimeMillis(); // 10.查询结束时间
Long nEndSelectTime = System.currentTimeMillis(); // 11.创建消息
javax.jms.TextMessage nMessage = nSession.createTextMessage();
nMessage.setText("我是发送的数据"); // 12.发送开始时间
Long nStartSendTime = System.currentTimeMillis(); // 13.生产者发布消息
nProducer.send(nMessage); // 14.消息结束时间
Long nEndSendTime = System.currentTimeMillis(); // 15.打印消息
System.out.println("已发送消息 "+nMessage); // 16.关闭发布者、会话、连接
nProducer.close();
nSession.close();
nConnection.close();
// 17.计算时间
System.out.println("查询消息时间 "+(nEndSelectTime-nStartSelectTime));
System.out.println("发送消息时间"+(nEndSendTime-nStartSendTime));
System.out.println("总时间"+(nEndSendTime-nStartTime));
} }

消息消费者


public class ZMQReadTest {
public static void main(String[] args) throws javax.jms.JMSException {
// 1.建立连接工厂
org.apache.activemq.ActiveMQConnectionFactory nFactory = new org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616"); // 2.用工厂建立Connection连接
javax.jms.Connection nTopicConnection = nFactory.createConnection(); // 3.创建会话Session,不启动事务,签收模式为自动确认模式
javax.jms.Session nSession = nTopicConnection.createSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE); // 4.创建队列
javax.jms.Destination nDestination = nSession.createQueue("123"); // 5.创建消费者
javax.jms.MessageConsumer nConsumer = nSession.createConsumer(nDestination);
// 6.为消费者添加消息监听器,onMessage方法接收消息
nConsumer.setMessageListener(new javax.jms.MessageListener() {
public void onMessage(javax.jms.Message nMessage) {
try {
if (nMessage instanceof javax.jms.TextMessage) {
String nTextMessage = ((javax.jms.TextMessage) nMessage).getText();
System.out.println("收到的消息:" + nTextMessage);
}
} catch (javax.jms.JMSException e) {
e.printStackTrace();
}
}
}); // 7.开启连接
nTopicConnection.start();
}
}