ActiveMQ消息队列点对点模式实例
点对点消息队列实例
注意 是点对点,不是订阅,就是一个生产者和一个消费者一一对应的,一个消费者接收了一条消息,其他人就收不到了;
- 下载activeMQ,windows版本,linux同理,启动就行了。
下载地址:http://activemq.apache.org/components/classic/download/
下载完成解压,进入bin目录下,进入32 位或者64位文件夹,双击activemq.bat执行文件,启动activeMQ.
输入网址:localhost:8161/admin/queues.jsp
网页弹出窗口,输入账号密码,都是admin
2. 创建发送消息的方法,说明都写在注释中
package com.ainy.demo;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
*
* <p>TODO 测试</p>
*
* @author LJW
* @version 2019年4月12日
*/
public class TestSendActiveMq {
public static void main(String[] args) throws Exception {
sendMessage("这是我的测试消息");
}
public static void sendMessage(String messageStr) throws Exception{
// 1.创建一个连接工厂对象ConnectionFactory对象。需要指定mq服务的ip及端口号。注意参数brokerURL的开头是
// tcp://而不是我们通常的http://,端口是61616而不是我们访问activemq后台管理页面所使用的8161
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
"tcp://10.2.7.7:61616");
// 2.使用ConnectionFactory创建一个连接Connection对象
Connection connection = connectionFactory.createConnection();
// 3.开启连接。调用Connection对象的start方法
connection.start();
// 4.使用Connection对象创建一个Session对象
// 第一个参数是是否开启事务,一般不使用分布式事务,因为它特别消耗性能,而且顾客体验特别差,现在互联网的
// 做法是保证数据的最终一致(也就是允许暂时数据不一致),比如顾客下单购买东西,一旦订单生成完就立刻响应给用户
// 下单成功。至于下单后一系列的操作,比如通知会计记账、通知物流发货、商品数量同步等等都先不用管,只需要
// 发送一条消息到消息队列,消息队列来告知各模块进行相应的操作,一次告知不行就两次,直到完成所有相关操作为止,这
// 也就做到了数据的最终一致性。如果第一个参数为true,那么第二个参数将会被忽略掉。如果第一个参数为false,那么
// 第二个参数为消息的应答模式,常见的有手动和自动两种模式,我们一般使用自动模式。
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
// 5.使用Session对象创建一个Destination对象,两种形式queue、topic。现在我们使用queue
// 参数就是消息队列的名称
Queue queue = session.createQueue("test-queue");
// 6.使用Session对象创建一个Producer对象
MessageProducer producer = session.createProducer(queue);
// 7.创建一个TextMessage对象
// 有两种方式,第一种方式:
// TextMessage textMessage = new ActiveMQTextMessage();
// textMessage.setText(messageStr);
// 第二种方式:
TextMessage textMessage = session.createTextMessage(messageStr);
// 8.发送消息
producer.send(textMessage);
// 9.关闭资源
producer.close();
session.close();
connection.close();
}
}
执行main方法。
点击消息队列名称:
点击消息ID:
在消息详情中可以看到我们刚才发送的消息
3. 接收消息
package com.ainy.demo;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
/**
* 接收消息
* @author LJW
* @version 2019年4月12日
*/
public class TestReceiveActiveMq {
public static void main(String[] args) throws Exception {
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://10.2.7.7:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("test-queue");
MessageConsumer consumer = session.createConsumer(queue);
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
String text;
try {
text = textMessage.getText();
System.out.println("接收到的消息:"+text);
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
});
// 8.程序等待接收用户结束操作
/*
* 程序自己不知道什么时候有消息,也不知道什么时候不再发送消息了,这就需要手动干预
* 当我们想停止接收消息时,可以在控制台输入任意键,然后回车(也可以直接按回车)即可结束接收操作
*/
System.in.read();
// 9.关闭资源
consumer.close();
session.close();
connection.close();
}
}
执行main方法,执行完 监听器会一直监听消息队列的消息,有一个 就接收一个。
,然后可以继续执行发送消息的方法,发送一次消息,消费端就会接收一次消息。