使用Java编写ActiveMQ的队列模式和主题模式
队列模式的消息演示
本小节简单演示一下如何使用JMS接口规范连接ActiveMQ,首先创建一个Maven工程,在pom.xml文件中,添加activemq的依赖:
<dependencies>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-all</artifactId>
<version>5.9.0</version>
</dependency>
</dependencies>
创建一个 AppProducer 类,用于演示下如何使用JMS接口规范使用ActiveMQ的队列模式。代码如下:
package org.zero01.jms.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @program: jms-test
* @description: 消息生产者
* @author: 01
* @create: 2018-05-26 16:44
**/
public class AppProducer {
// activemq服务器的url地址,默认通信端口为61616
private static final String URL = "tcp://192.168.190.129:61616";
// 队列的名称
private static final String QUEUE_NAME = "queue-test";
public static void main(String[] args) throws JMSException {
// 1.创建连接工厂对象(ConnectionFactory)
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
// 2.创建连接对象(Connection)
Connection connection = connectionFactory.createConnection();
// 3.启动连接
connection.start();
// 4.创建session会话,第一参数表示启用事务处理,第二个参数表示启动哪种应答模式,这里启用的是自动应答
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 5.创建目的地(destination)
Destination destination = session.createQueue(QUEUE_NAME);
// 6.创建生产者
MessageProducer producer = session.createProducer(destination);
// 循环发送消息
for (int i = 0; i < 100; i++) {
// 7.创建消息,这里创建的是简单的文本消息体
TextMessage textMessage = session.createTextMessage("test" + i);
// 8.使用消息生产者往目的地发送消息
producer.send(destination, textMessage);
System.out.println("消息发送成功:" + textMessage.getText());
}
// 9.关闭连接
connection.close();
}
}
编写完代码后,登录ActiveMQ的管理页面,点击选项卡上的 “Queues” 进入到如下界面,可以看到现在这里什么数据都没有:
我们运行上面编写的代码之后,刷新该页面,可以看到现在就有数据了:
接着我们来编写一个消费者,去消费队列中的消息。代码如下:
package org.zero01.jms.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @program: jms-test
* @description: 消息消费者
* @author: 01
* @create: 2018-05-26 17:08
**/
public class AppConsumer {
// activemq服务器的url地址,默认通信端口为61616
private static final String URL = "tcp://192.168.190.129:61616";
// 队列的名称
private static final String QUEUE_NAME = "queue-test";
public static void main(String[] args) throws JMSException {
// 1.创建连接工厂对象(ConnectionFactory)
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
// 2.创建连接对象(Connection)
Connection connection = connectionFactory.createConnection();
// 3.启动连接
connection.start();
// 4.创建session会话,第一参数表示启用事务处理,第二个参数表示启动哪种应答模式,这里启用的是自动应答
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 5.创建目的地(destination)
Destination destination = session.createQueue(QUEUE_NAME);
// 6.创建消费者
MessageConsumer messageConsumer = session.createConsumer(destination);
// 7.创建一个监听器
messageConsumer.setMessageListener((message) -> {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("接收消息: " + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
});
}
}
编写完代码后,点击选项卡上的 “Connections” 进入到如下界面,可以看到现在这里一个连接都没有:
我们运行上面编写的代码之后,刷新该页面,可以看到现在就有一个消费者连接了:
消费者运行之后,是一个线程阻塞状态的,也就是会与ActiveMQ服务器保持连接。现在我们再来启动一个消费者,如下就有两个消费者了:
启动了两个消费者后,运行生产者的代码。我们来看队列模式的一个现象,如下:
控制台打印信息如上,有没有发现,消费者1所消费的消息是偶数的,而消费者2消费的消息则是奇数的。这就是队列模式的一个现象,消费者们会均匀地、尽可能平均地消费队列中的消息。
主题模式的消息演示
主题模式的代码和队列模式的代码十分类似,只有创建目的地的方法不一样。消息发布者代码如下:
package org.zero01.jms.topic;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @program: jms-test
* @description: 消息发布者
* @author: 01
* @create: 2018-05-26 16:44
**/
public class AppPublisher {
// activemq服务器的url地址,默认通信端口为61616
private static final String URL = "tcp://192.168.190.129:61616";
// 队列的名称
private static final String TOPIC_NAME = "topic-test";
public static void main(String[] args) throws JMSException {
// 1.创建连接工厂对象(ConnectionFactory)
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
// 2.创建连接对象(Connection)
Connection connection = connectionFactory.createConnection();
// 3.启动连接
connection.start();
// 4.创建session会话,第一参数表示启用事务处理,第二个参数表示启动哪种应答模式,这里启用的是自动应答
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 5.创建目的地(destination)
Destination destination = session.createTopic(TOPIC_NAME);
// 6.创建生产者
MessageProducer producer = session.createProducer(destination);
// 循环发送消息
for (int i = 0; i < 100; i++) {
// 7.创建消息,这里创建的是简单的文本消息体
TextMessage textMessage = session.createTextMessage("test" + i);
// 8.使用消息生产者往目的地发送消息
producer.send(destination, textMessage);
System.out.println("消息发送成功:" + textMessage.getText());
}
// 9.关闭连接
connection.close();
}
}
消息订阅者代码如下:
package org.zero01.jms.topic;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* @program: jms-test
* @description: 消息订阅者
* @author: 01
* @create: 2018-05-26 17:08
**/
public class AppSubscriber {
// activemq服务器的url地址,默认通信端口为61616
private static final String URL = "tcp://192.168.190.129:61616";
// 队列的名称
private static final String TOPIC_NAME = "topic-test";
public static void main(String[] args) throws JMSException {
// 1.创建连接工厂对象(ConnectionFactory)
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
// 2.创建连接对象(Connection)
Connection connection = connectionFactory.createConnection();
// 3.启动连接
connection.start();
// 4.创建session会话,第一参数表示启用事务处理,第二个参数表示启动哪种应答模式,这里启用的是自动应答
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 5.创建目的地(destination)
Destination destination = session.createTopic(TOPIC_NAME);
// 6.创建消费者
MessageConsumer messageConsumer = session.createConsumer(destination);
// 7.创建一个监听器
messageConsumer.setMessageListener((message) -> {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println("接收消息: " + textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
});
}
}
但主题模式与队列模式的消费方式不太一样,队列模式是先启动消息生产者去发送消息到队列里,然后消费者再去消费。而主题模式则是先启动消息订阅者去进行订阅,然后再启动消息发布者去发布消息,这样消息订阅者才能收到消息发布者所发布的消息。所以我们先启动消息订阅者,再启动消息发布者。完成启动后,这时到ActiveMQ的 “Topics” 页面上查看信息如下:
除了以上这个区别外,我们来启动两个订阅者,然后再启动发布者,看看订阅者们接收到的消息是怎么样的:
控制台打印信息如上,可以看到两个订阅者都各自接收到了同一份消息。也就是说,如果有两个订阅者,那么消息就会有两份,有多个订阅者则有多份。
- 算法模板——splay区间反转 2
- 算法模板——Dinic网络最大流 2
- 1935: [Shoi2007]Tree 园丁的烦恼
- 1339 / 1163: [Baltic2008]Mafia
- 4010: [HNOI2015]菜肴制作
- 4052: [Cerc2013]Magical GCD
- ElasticSearch搜索引擎在SpringBoot中的实践
- 2292: 【POJ Challenge 】永远挑战
- 四边形不等式优化DP
- 4063: [Cerc2012]Darts
- 3997: [TJOI2015]组合数学
- 1349: [Baltic2006]Squint
- 遗传算法(2)
- 算法模板——线性筛素数
- java教程
- Java快速入门
- Java 开发环境配置
- Java基本语法
- Java 对象和类
- Java 基本数据类型
- Java 变量类型
- Java 修饰符
- Java 运算符
- Java 循环结构
- Java 分支结构
- Java Number类
- Java Character类
- Java String类
- Java StringBuffer和StringBuilder类
- Java 数组
- Java 日期时间
- Java 正则表达式
- Java 方法
- Java 流(Stream)、文件(File)和IO
- Java 异常处理
- Java 继承
- Java 重写(Override)与重载(Overload)
- Java 多态
- Java 抽象类
- Java 封装
- Java 接口
- Java 包(package)
- Java 数据结构
- Java 集合框架
- Java 泛型
- Java 序列化
- Java 网络编程
- Java 发送邮件
- Java 多线程编程
- Java Applet基础
- Java 文档注释