一、什么是?
JMS : Java Message Service(Java消息服务),Java平台中关于面向消息中间件的接口
重点在于接口,接口就意味着与JDBC类似,仅仅有声明,没有实现,具体的实现交给厂商. 接口本身是一种与厂商无关的API.
支持的消息类型
TextMessage 字符串对象
MapMessage 名值对
ObjectMessage 一个序列化的java对象
BytesMessage 字节的数据流
StreamMessage 原始数据值的数据流
作用
优点 使用消息服务器 当做大一点的队列使用 先进先出 处理高并发写入问题，似的=业务由串行改为并行 处理效率高
缺点：实时性不高 因为数据发送给消息服务器 消息服务器不会立刻处理 队列是先进先出，如果数据量大需要排队等候
二、使用场景
每一种技术都有其产生的原因和解决的问题,消息服务的使用场景如下:
a:大的队列使用 先进先出 来处理高并发问题
b:业务系统的串行改为并行 处理效率高
- 核心应用:
- 解耦: 订单 --> 物流 (降低工程间的强依赖程度)
- 异步: 用户注册 --> 发送邮件,初始化信息 (耗时长非必须的业务抽离,放在空闲的时候处理)
- 流量削峰: 秒杀,日志处理 (针对突发情况的大访问量场景)
- 跨平台, 多语言
- 分布式事务, 最终一致性 (引入和分布式和微服务之后,事务的处理和数据的一致是比较难保证的一个方面)
- RPC调用上下游对接,数据源变动通知下属
三、消息中间件常见概念与编程模型
常见概念
- JMS提供者: 连接面向消息中间件的,也就是JMS的接口实现, RocketMQ, ActiveMQ等. 简单的来说就是JMS接口的实现厂商提供的实现, 中间件本身.
- Broker服务器: 接收生产消息,提供给消费者消费的程序
- JMS生产者(Message Producer): 生产消息的服务
- JMS消费者(Message Consumer): 消费消息的服务
- JMS消息: 数据对象, 传递的内容
- JMS队列: 存储待消费信息的区域(可以简单的理解为数据结构中的队列,可以保证顺序,存储消息的介质)
- JMS主题(Topic): 一种支持发送消息给多个订阅者的机制(中转,不同的生产者发送消息到主题,经过服务器分发到不同的订阅者)
- JMS消息通常的两种类型: 点对点(Point-to-Point), 发布/订阅(Publish/Subscribe). 区别在于前者是多个消费者但只有一个消费者可以消费消息,后者是多个消费者都可以同时消费一个消息.
一个消息对应一个主题,一个主题下由多个queue. 要注意的是topic是一个逻辑管理单位,queue是物理管理单位
基础编程模型
- MQ中需要用到的一些类
- ConnectionFactory: 连接工厂, JMS使用它来创建连接
- Connection: JMS客户端到JMS Provider的连接
- Session: 一个发送或接收消息的线程
- Destination: 消息的目的地; 消息发送给谁(Broker)
- MessageConsumer / MessageProducer: 消息消费者 / 消息生产者
四、同类技术对比
- 开源的AMQP实现,服务端用Erlang语言编写,支持多种客户端,在易用性,扩展性和高可用性等方面表现不错. 但是使用Erlang语言开发,阅读和修改源码难度大.(当然公司如果大部分人熟悉Erlang语言的话例外)安全性高
- ZeroMQ: (最快)
五、小Demo
pom.xml
<dependencies>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<version>5.13.4</version>
</dependency>
</dependencies>
点对点模式
接收方
package cn.liuhuan.core;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 接收方
*/
public class QueueConsuer {
public static void main(String[] args) throws Exception {
//1 创建连接工厂
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.200.128:61616");
//2 获得链接
Connection connection = connectionFactory.createConnection();
//3 启动连接
connection.start();
//4 获取session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5 创建队列
Queue queue = session.createQueue("test01");
//6 创建消息
MessageConsumer consumer = session.createConsumer(queue);
//7 监听消息
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println(textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//8 等待键盘输入
System.in.read();
//9 关闭资源
consumer.close();
session.close();
connection.close();
}
}
发送方
package cn.liuhuan.core;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 发送方
*/
public class QueueProducer {
public static void main(String[] args) throws Exception {
//1 创建连接工厂
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.200.128:61616");
//2 获得链接
Connection connection = connectionFactory.createConnection();
//3 启动连接
connection.start();
//4 获取session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5 创建队列
Queue queue = session.createQueue("test01");
//6 创建消息
MessageProducer producer = session.createProducer(queue);
//7 生产消息
TextMessage textMessage = session.createTextMessage("这是收到的第一个消息");
//8 发送消息
producer.send(textMessage);
//9 关闭资源
producer.close();
session.close();
connection.close();
}
}
订阅发布模式
接收方(多个都是一样的)
package cn.liuhuan.core;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class TopicConsuer1 {
public static void main(String[] args) throws Exception {
//1 创建连接工厂
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.200.128:61616");
//2 获得链接
Connection connection = connectionFactory.createConnection();
//3 启动连接
connection.start();
//4 获取session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5 创建队列
Topic topic = session.createTopic("testTopic");
//6 创建消息
MessageConsumer consumer = session.createConsumer(topic);
//7 监听消息
consumer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage) message;
try {
System.out.println(textMessage.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
});
//8 等待键盘输入
System.in.read();
//9 关闭资源
consumer.close();
session.close();
connection.close();
}
}
发送方
package cn.liuhuan.core;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 发送方
*/
public class TopicProducer {
public static void main(String[] args) throws Exception {
//1 创建连接工厂
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.200.128:61616");
//2 获得链接
Connection connection = connectionFactory.createConnection();
//3 启动连接
connection.start();
//4 获取session
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
//5 创建队列
Topic topic = session.createTopic("testTopic");
//6 创建消息
MessageProducer producer = session.createProducer(topic);
//7 生产消息
TextMessage textMessage = session.createTextMessage("这是收到的第一个消息");
//8 发送消息
producer.send(textMessage);
//9 关闭资源
producer.close();
session.close();
connection.close();
}
}