|
|
|
步骤 2 : 先运行，看到效果，再学习 步骤 3 : 模仿和排错 步骤 4 : 管理界面 步骤 5 : pom.xml 步骤 6 : ActiveMQUtil 步骤 7 : TestProducer 步骤 8 : TestConsumer
activeMQ 有两种模式，分别是队列模式和主题模式。
队列模式，其实就是分食模式。 比如生产方发了 10条消息到 activeMQ 服务器， 而此时有多个 消费方，那么这些消费方就会瓜分这些10条消息，一条消息只会被一个消费方得到。 主题模式，就是订阅模式。 比如生产方发了10条消息，而此时有多个消费方，那么多个消费方都能得到这 10条消息，就如同订阅公众号那样。 而本知识点会演示如何开发队列模式。
老规矩，先下载右上角的可运行项目，配置运行起来，确认可用之后，再学习做了哪些步骤以达到这样的效果。
这是个 maven 项目，不熟悉 maven的同学可学习本站的 maven基础教程： 什么是Maven 测试办法： 1. 首先运行两次 TestConsumer 类，以启动两个不同的消费者 2. 运行一次 TestProducer， 以启动 生产者。 这个生产者会生产100条消息。 这两个消费者会分食这100条消息，如图所示这样的效果，两个消费者分别瓜分不同的消息。
在确保可运行项目能够正确无误地运行之后，再严格照着教程的步骤，对代码模仿一遍。
模仿过程难免代码有出入，导致无法得到期望的运行结果，此时此刻通过比较正确答案 ( 可运行项目 ) 和自己的代码，来定位问题所在。 采用这种方式，学习有效果，排错有效率，可以较为明显地提升学习速度，跨过学习路上的各个槛。 推荐使用diffmerge软件，进行文件夹比较。把你自己做的项目文件夹，和我的可运行项目文件夹进行比较。 这个软件很牛逼的，可以知道文件夹里哪两个文件不对，并且很明显地标记出来 这里提供了绿色安装和使用教程：diffmerge 下载和使用教程
访问地址： http://127.0.0.1:8161/admin/queues.jsp 就可以看到 刚才的消息处理情况。
queue_style 是在代码中定义的消息名称。 number Of Consumers 表示有2个消费者。 Messages Enqueued：表示收到了 100 个消息。 Messages Dequeued：表示消费了 100 个消息。
导入两个包，一个是 activemq ，另一个是 hutool jar包
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>cn.how2j</groupId>
<artifactId>activemq</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>activemq</name>
<description>activemq</description>
<dependencies>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.15.9</version>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>4.3.1</version>
</dependency>
</dependencies>
</project>
工具类，用于判断 activemq 服务器是否启动，免得有些学员在服务器没有启动的情况先，搞半天都搞不出来。
package cn.how2j.util;
import javax.swing.JOptionPane;
import cn.hutool.core.util.NetUtil;
public class ActiveMQUtil {
public static void main(String[] args) {
checkServer();
}
public static void checkServer() {
if(NetUtil.isUsableLocalPort(8161)) {
JOptionPane.showMessageDialog(null, "ActiveMQ 服务器未启动 ");
System.exit(1);
}
}
}
package cn.how2j.util;
import javax.swing.JOptionPane;
import cn.hutool.core.util.NetUtil;
public class ActiveMQUtil {
public static void main(String[] args) {
checkServer();
}
public static void checkServer() {
if(NetUtil.isUsableLocalPort(8161)) {
JOptionPane.showMessageDialog(null, "ActiveMQ 服务器未启动 ");
System.exit(1);
}
}
}
生产100条消息。
每行都做了注释，直接看代码注释就理解了，这里就不赘述了。 package cn.how2j.queue;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
import cn.how2j.util.ActiveMQUtil;
public class TestProducer {
//服务地址，端口默认61616
private static final String url="tcp://127.0.0.1:61616";
//这次发送的消息名称
private static final String topicName="queue_style";
public static void main(String[] args) throws JMSException {
//0. 先判断端口是否启动了 Active MQ 服务器
ActiveMQUtil.checkServer();
//1.创建ConnectionFactory,绑定地址
ConnectionFactory factory=new ActiveMQConnectionFactory(url);
//2.创建Connection
Connection connection= factory.createConnection();
//3.启动连接
connection.start();
//4.创建会话
Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5.创建一个目标 (队列类型)
Destination destination=session.createQueue(topicName);
//6.创建一个生产者
MessageProducer producer=session.createProducer(destination);
for (int i = 0; i < 100; i++) {
//7.创建消息
TextMessage textMessage=session.createTextMessage("队列消息-"+i);
//8.发送消息
producer.send(textMessage);
System.out.println("发送："+textMessage.getText());
}
//7. 关闭连接
connection.close();
}
}
消费者，消费服务器上的消息。
同样的也对每行都做了注释，仔细阅读便能理解。 package cn.how2j.queue;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
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 org.apache.activemq.ActiveMQConnectionFactory;
import cn.how2j.util.ActiveMQUtil;
import cn.hutool.core.util.RandomUtil;
/**
* 订阅者
* @author root
*
*/
public class TestConsumer {
//服务地址，端口默认61616
private static final String url="tcp://127.0.0.1:61616";
//这次消费的消息名称
private static final String topicName="queue_style";
//消费者有可能是多个，为了区分不同的消费者，为其创建随机名称
private static final String consumerName="consumer-" + RandomUtil.randomString(5);
public static void main(String[] args) throws JMSException {
//0. 先判断端口是否启动了 Active MQ 服务器
ActiveMQUtil.checkServer();
System.out.printf("%s 消费者启动了。 %n", consumerName);
//1.创建ConnectiongFactory,绑定地址
ConnectionFactory factory=new ActiveMQConnectionFactory(url);
//2.创建Connection
Connection connection= factory.createConnection();
//3.启动连接
connection.start();
//4.创建会话
Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5.创建一个目标 （队列类型）
Destination destination=session.createQueue(topicName);
//6.创建一个消费者
MessageConsumer consumer=session.createConsumer(destination);
//7.创建一个监听器
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message arg0) {
// TODO Auto-generated method stub
TextMessage textMessage=(TextMessage)arg0;
try {
System.out.println(consumerName +" 接收消息："+textMessage.getText());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
//8. 因为不知道什么时候有，所以没法主动关闭，就不关闭了，一直处于监听状态
//connection.close();
}
}
HOW2J公众号，关注后实时获知最新的教程和优惠活动，谢谢。
问答区域
2021-06-30
消费者按类或main方法start两次都只生成一个消费者?
2020-07-30
找不到jms的包
2020-04-08
Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 了解第一个第二个参数
2019-07-20
mac中无法通过checkServer检测, 但是可以运行其他代码
2019-07-03
为什么我只能在控制台看到发送队列消息啊
提问太多，页面渲染太慢，为了加快渲染速度，本页最多只显示几条提问。还有 2 条以前的提问，请 点击查看
提问之前请登陆
提问已经提交成功，正在审核。 请于 我的提问 处查看提问记录，谢谢
|
|||||||||
