ActiveMQ学习之java代码调用ActiveMQ队列
ActiveMQ学习之java代码调用ActiveMQ队列
用户5899361
发布于 2020-12-07 15:07:54
发布于 2020-12-07 15:07:54
代码可运行
运行总次数：0
代码可运行
一、基本概念：
ActiveMQ中共分为两种：queue和topic
queue：在点对点消息传递域中，目的地被称为队列（一对一）
topic：在发布订阅消息中，目的地被称为主题（一对多）
二、创建maven工程，并引入依赖，这里我创建的springboot项目，所以引入的依赖如下：
依赖：
<!--activemq-->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.9</version>
</dependency>创建main方法测试连接，消息提供者。
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @ProjectName: springbootActiveMQ
* @Package: cn.**.test
* @Author: huat
* @Date: 2020/1/2 17:04
* @Version: 1.0
*/
public class ActiveMQTest {
//url路径
private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
//队列名称
private static final String QUEUE_NAME="queue01";
public static void main(String[] args) {
//1、创建连接工厂
//如果账号密码没有修改的话，账号密码默认均为admin
ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
//如果账号密码修改的话
//第一个参数为账号，第二个为密码，第三个为请求的url
//ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
try {
//2、通过连接工厂获取连接
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3、创建session会话
//里面会有两个参数，第一个为事物，第二个是签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4、创建目的地（具体是队列还是主题）,这里是创建队列
Queue queue=session.createQueue(QUEUE_NAME);
//5、创建消息生产者
MessageProducer messageProducer = session.createProducer(queue);
//6、通过messageProducer生产三条消息发送到MQ消息队列中
for (int i=0;i<3;i++){
//7、创建消息
TextMessage textMessage = session.createTextMessage("msg----->" + i);//创建一个文本消息
//8、通过messageProducer发送给mq
messageProducer.send(textMessage);
}
messageProducer.close();
session.close();
connection.close();
System.out.println("消息发送成功");
} catch (JMSException e) {
e.printStackTrace();
}
}
}查看管理页面是否有消息
创建main方法测试连接，消息消费者。（同步调用）
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @ProjectName: springbootActiveMQ
* @Package: cn.**.test
* @Author: huat
* @Date: 2020/1/3 8:47
* @Version: 1.0
*/
public class ActiveMQConsumer {
//url路径
private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
//队列名称
private static final String QUEUE_NAME="queue01";
public static void main(String[] args) {
//1、创建连接工厂
//如果账号密码没有修改的话，账号密码默认均为admin
ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
//如果账号密码修改的话
//第一个参数为账号，第二个为密码，第三个为请求的url
//ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
try {
//2、通过连接工厂获取连接
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3、创建session会话
//里面会有两个参数，第一个为事物，第二个是签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4、这里接受的queue的名称要和发送者的一致
Queue queue = session.createQueue(QUEUE_NAME);
//5、创建消费者
MessageConsumer consumer = session.createConsumer(queue);
//6、获取消息
while(true){
//MessageConsumer 调用的receive方法为同步调用，在消息到达之前一直阻塞线程
//用什么格式发送，这里就用什么格式接受
//receive等待消息，不限制时间
TextMessage message=(TextMessage)consumer.receive();
//receive带参数等待消息，限制时间，单位毫秒
//TextMessage message=(TextMessage)consumer.receive(4000L);
if(null != message){
System.out.println("接受的消息为------>"+message.getText());
}else{
break;
}
}
//7、闭资源
consumer.close();
session.close();
connection.close();
}catch (Exception e){
e.printStackTrace();
}
}
}创建main方法测试连接，消息消费者。（异步调用）
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @ProjectName: springbootActiveMQ
* @Package: cn.**.test
* @Author: huat
* @Date: 2020/1/3 8:47
* @Version: 1.0
*/
public class ActiveMQConsumer {
//url路径
private static final String ACTRIVE_URL="tcp://192.168.44.135:61616";
//队列名称
private static final String QUEUE_NAME="queue01";
public static void main(String[] args) {
//1、创建连接工厂
//如果账号密码没有修改的话，账号密码默认均为admin
ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ACTRIVE_URL);
//如果账号密码修改的话
//第一个参数为账号，第二个为密码，第三个为请求的url
//ActiveMQConnectionFactory activeMQConnectionFactory1=new ActiveMQConnectionFactory("admin","admin",ACTRIVE_URL);
try {
//2、通过连接工厂获取连接
Connection connection = activeMQConnectionFactory.createConnection();
connection.start();
//3、创建session会话
//里面会有两个参数，第一个为事物，第二个是签收
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//4、这里接受的queue的名称要和发送者的一致
Queue queue = session.createQueue(QUEUE_NAME);
//5、创建消费者
MessageConsumer consumer = session.createConsumer(queue);
//6、通过监听的方式消费消息
consumer.setMessageListener(new MessageListener() {
@Override
public void onMessage(Message message) {
//如果message不等于null并且属于TextMessage类型（因为消息发送的类型是TextMessage，所以这里判断是否是这个类型）
if(null!=message&&message instanceof TextMessage){
TextMessage textMessage=(TextMessage)message;
try {
System.out.println(textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}
});
//7、保证控制台一直在运行
System.in.read();
//8、闭资源
consumer.close();
session.close();
connection.close();
}catch (Exception e){
e.printStackTrace();
}
}
}
三、具体分析
情况一：先启动生产者，再启动一号消费者，一号消费者可以消费全部消息
情况二：先启动生产者，先启动一号消费者再启动二号消费者，一号消费者可以消费全部消息，二号消费者无消费
情况三：先启动一号和二号两个消费者，再启动生产者，一号和二号消费者平均分配消息。
本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。
如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
腾讯云开发者
