ActiveMQ的学习整理(代码实现PTP,以及Pub/Sub)
(一)由于在实习过程中需要用到ActiveMQ,在网上看了很多文章,现在整理出来以防忘记。
(二)这篇文章比较适合之前没有接触过的同学,在看下面文章的过程中,建议先学习参考链接中的知识点,然后自己再参考我的代码来实现实现PTP,以及Pub/Sub,两遍下来基本能搞定ActiveMQ的基础知识。
(三)参考链接:https://www.jianshu.com/p/b547db55d168
这是关于消息中间件ActiveMQ的一个系列专题文章,将涵盖JMS、ActiveMQ的初步入门及API详细使用、两种经典的消息模式(PTP and Pub/Sub)、与Spring整合、ActiveMQ集群、监控与配置优化等。
JMS
首先来说较早以前,也就是没有JMS的那个时候,很多应用系统存在一些缺陷:
1.通信的同步性:client端发起调用后,必须等待server处理完成并返回结果后才能继续执行;
2.client 和 server 的生命周期耦合太高:client进程和server服务进程都必须可用,如果server出现问题或者网络故障,那么client端会收到异常;
3.点对点通信:client端的一次调用只能发送给某一个单独的服务对象,无法一对多;
需要注意的是,JMS只是定义了Java访问消息中间件的接口,其实就是在包javax.jms中,你会发现这个包下除了异常定义,其他都是interface。我们可以扫一眼,比如Message:
JMS只给出接口,然后由具体的中间件去实现,比如ActiveMQ就是实现了JMS的一种Provider,还有阿里巴巴的RocketMQ(后续专题中在为大家介绍)。这些消息中间件都符合JMS规范。说起规范,自然要定义一些术语:
Provider/MessageProvider:生产者
Consumer/MessageConsumer:消费者
PTP:Point To Point,点对点通信消息模型
Pub/Sub:Publish/Subscribe,发布订阅消息模型
Queue:队列,目标类型之一,和PTP结合
Topic:主题,目标类型之一,和Pub/Sub结合
ConnectionFactory:连接工厂,JMS用它创建连接
Connnection:JMS Client到JMS Provider的连接
Destination:消息目的地,由Session创建
Session:会话,由Connection创建,实质上就是发送、接受消息的一个线程,因此生产者、消费者都是Session创建的
PTP模式实现
PTP模式特点:
1,每个消息只有一个消费者(Consumer)(即一旦被消费,消息就不再在消息队列中)
2,发送者和接收者之间在时间上没有依赖性,也就是说当发送者发送了消息之后,不管接收者有没有正在运行,它不会影响到消息被发送到队列
3,接收者在成功接收消息之后需向队列应答成功
(一)在IntelliJ IDEA中新建project,然后再建立两个项目,以及相应的类
(二)启动ActiveMQ
双击apache-activemq-5.5.1\bin\activemq.bat运行ActiveMQ程序。启动ActiveMQ以后,登陆:http://localhost:8161/admin/(默认账号密码为admin),创建一个Queue,命名为FirstQueue。
详细过程参考:https://www.cnblogs.com/xwdreamer/archive/2012/02/21/2360818.html
(三)sender与receiver的代码
sender:
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*;
import java.util.Date; public class sender {
private static final int SEND_NUMBER = 5; public static void main(String[] args) {
// ConnectionFactory :连接工厂,JMS 用它创建连接
ConnectionFactory connectionFactory;
// Connection :JMS 客户端到JMS Provider 的连接
Connection connection = null;
// Session: 一个发送或接收消息的线程
Session session;
// Destination :消息的目的地;消息发送给谁.
Destination destination;
// MessageProducer:消息发送者
MessageProducer producer; // TextMessage message;
// 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar
connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD,"tcp://localhost:61616");
try {
// 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操作连接
session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置
destination = session.createQueue("FirstQueue");
// 得到消息生成者【发送者】
producer = session.createProducer(destination);
// 设置不持久化,此处学习,实际根据项目决定
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 构造消息,此处写死,项目就是参数,或者方法获取
sendMessage(session, producer);
session.commit();
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection) {
connection.close();
}
} catch (Throwable ignore) {
}
}
} public static void sendMessage(Session session, MessageProducer producer) throws Exception {
for (int i = 1; i <= SEND_NUMBER; i++) {
Date date = new Date();
TextMessage message = session.createTextMessage(date + " ActiveMQ 发送的消息:" + i);
// 发送消息到目的地方
System.out.println("发送消息: " + date + " ActiveMQ 发送的消息 " + i);
producer.send(message);
}
}
}
receiver:
import org.apache.activemq.*;
import javax.jms.*; public class receiver {
public static void main(String[] args) {
// ConnectionFactory :连接工厂,JMS 用它创建连接
ConnectionFactory connectionFactory;
// Connection :JMS 客户端到JMS Provider 的连接
Connection connection = null;
// Session: 一个发送或接收消息的线程
Session session;
// Destination :消息的目的地;消息发送给谁.
Destination destination;
// 消费者,消息接收者
MessageConsumer consumer;
connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616");
try {
// 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操作连接
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置
destination = session.createQueue("FirstQueue");
consumer = session.createConsumer(destination);
while (true) {
//设置接收者接收消息的时间,为了便于测试,这里谁定为100s
TextMessage message = (TextMessage) consumer.receive(100000);
if (null != message) {
System.out.println("收到消息:" + message.getText());
} else {
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
if (null != connection){
connection.close();
}
} catch (Throwable ignore) {
}
}
}
}
(四)调试效果
Pub/Sub模式
Pub/Sub模式特点
1,每个消息可以有多个消费者
2,发布者和订阅者之间有时间上的依赖性。针对某个主题(Topic)的订阅者,它必须创建一个订阅者之后,才能消费发布者的消息,而且为了消费消息,订阅者必须保持运行的状态。
3,为了缓和这样严格的时间相关性,JMS允许订阅者创建一个可持久化的订阅。这样,即使订阅者没有被激活(运行),它也能接收到发布者的消息。
(一)新建项目
(二)show the code
TopicProducer:
package pub_sub; import org.apache.activemq.*;
import javax.jms.*;
import java.util.Date; public class TopicProducer {
private static final int SEND_NUMBER = 5;
public static void main(String[] args) {
// ConnectionFactory :连接工厂,JMS 用它创建连接
ConnectionFactory connectionFactory;
// Connection :JMS 客户端到JMS Provider 的连接
Connection connection = null;
// Session: 一个发送或接收消息的线程
Session session;
// Destination :消息的目的地;消息发送给谁.
Destination destination;
// MessageProducer:消息发送者
MessageProducer producer;
// TextMessage message;
// 构造ConnectionFactory实例对象,此处采用ActiveMq的实现jar
connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD,"tcp://localhost:61616"); try {
// 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操作连接
session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 获取session注意参数值是一个服务器的Topic,须在在ActiveMq的console配置
destination = session.createTopic("FirstTopic");
producer = session.createProducer(destination);
// 设置不持久化,此处学习,实际根据项目决定
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
// 构造消息,此处写死,项目就是参数,或者方法获取
sendMessage(session, producer);
session.commit(); } catch (Exception e) {
e.printStackTrace();
}
finally {
try {
if (null != connection) {
connection.close();
}
} catch (Throwable ignore) {
}
}
} public static void sendMessage(Session session, MessageProducer producer) throws Exception {
for (int i = 1; i <= SEND_NUMBER; i++) {
Date date = new Date();
TextMessage message = session.createTextMessage(date + " ActiveMQ 发送的消息:" + i);
// 发送消息到目的地方
System.out.println("发送消息: " + date + " ActiveMQ 发送的消息 " + i);
producer.send(message);
}
}
}
TopicConsumer:
package receiver; import org.apache.activemq.*;
import javax.jms.*; public class TopicConsumer{ public static void main(String[] args) {
// ConnectionFactory :连接工厂,JMS 用它创建连接
ConnectionFactory connectionFactory;
// Connection :JMS 客户端到JMS Provider 的连接
Connection connection = null;
// Session: 一个发送或接收消息的线程
Session session;
// Destination :消息的目的地;消息发送给谁.
Destination destination;
// 消费者,消息接收者
MessageConsumer consumer;
connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://localhost:61616"); try {
// 构造从工厂得到连接对象
connection = connectionFactory.createConnection();
// 启动
connection.start();
// 获取操作连接
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// 获取session注意参数值是一个服务器的Topic,须在在ActiveMq的console配置
destination = session.createTopic("FirstTopic");
consumer = session.createConsumer(destination);
// while (true) {
// //设置接收者接收消息的时间,为了便于测试,这里谁定为100s
// TextMessage message = (TextMessage) consumer.receive(100000);
// if (null != message) {
// System.out.println("收到消息:" + message.getText());
// } else {
// break;
// }
// }
while(true){
consumer.setMessageListener(new listern());
}
} catch (Exception e) {
e.printStackTrace();
}
finally {
try {
if (null != connection){
connection.close();
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
listern:
package receiver; import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage; public class listern implements MessageListener {
@Override
public void onMessage(Message message) {
try {
System.out.println("收到的消息:"+((TextMessage)message).getText());
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
(三)注意事项
1. 运行代码的时候要先运行订阅者代码,然后运行发布者代码。
2. Consumer中注释代码为不添加Listern的实现。
最新文章
- knockoutJS学习笔记08:表单域绑定
- RN中的onChangeText
- Exel 利用模板导出方法
- oracle数据库表空间及归档
- Atmospheric Scattering in Unity5
- [待解决问题] 启动不了Android工程
- 关于Oracle dmp文件导入随笔
- SVG动画实践篇-模拟音量高低效果
- TCP/IP和UDP之间的区别(转载)
- Python读取配置文件,并连接数据库SQL Server
- Spring Security入门(2-2)Spring Security 的运行原理 2
- Java复数的加乘除运算
- Java 接口 Comparable
- A Simple Note on ";P4FPGA: A Rapid Prototyping Framework for P4";
- Trailing Zeroes (I) LightOJ - 1028
- TCP/IP 笔记 - 概述
- vue---进行post和get请求
- Centos6.8/7.0搭建Git服务http访问方式
- sublime package control失败
- jQuery左侧菜单实例
热门文章
- oracle 之 插入超长字段并包含&;字符的处理方法
- kubernetes 实战3_命令_Configure Pods and Containers
- 使用p4c将P4 14代码转换为16代码
- vim命令详解
- trackViewer 氨基酸位点变异位置图谱展示
- Codeforces Round #135 (Div. 2) D. Choosing Capital for Treeland dfs
- SPOJ QTREE Query on a tree 树链剖分+线段树
- idea创建web聚合工程(2)
- MySql连接时出现1251 client does no support authentic错误解决方法
- Eclipse+maven 构建第一个简单的springmvc项目