消息中间件

介绍

什么是JMS：
jms的全称是Java message service（Java消息服务），jms是jdk底层定义的规范，各大厂商都有实现这个规范的技术。

作用：在soa分布式架构系统中，或企业的多个项目中，进行多个系统异步传递消息。

同步和异步：

• 同步技术:
  dubbo是一中同步技术, 实时性高, controller调用service项目, 调用就执行, 如果service项目中的代码没有执行完, controller里面的代码一致等待结果
• 异步技术:
  mq消息中间件技术(jms) 是一种异步技术, 消息发送方, 将消息发送给消息服务器, 消息服务器未必立即处理.什么时候去处理, 主要看消息服务器是否繁忙, 消息进入服务器后会进入队列中, 先进先出.实时性不高.

使用场景：
场景一：使用消息服务器当做大的队列使用, 先进先出, 来处理高并发写入操作

场景二：使用消息服务器可以将业务系统的串行执行改为并行执行, 处理效率高, 更合理的榨取服务器的性能.

缺点:
消息中间件实时性不高, 因为发送方将消息发送给消息服务器后, 消息服务器未必能够立即处理, 主要看消息服务器是否繁忙, 队列中先进先出, 如果数据量大, 要排队等待执行

jms消息服务器同类型技术:

• ActiveMQ:
  是apache的一个比较老牌的消息中间件, 它比较均衡, 既不是最安全的, 也不是最快的.
• RabbitMQ:
  是阿里巴巴的一个消息中间件, 更适合金融类业务, 它对数据的安全性比较高.能够保证数据不丢失.
• ZeroMQ:
  史上最快的消息队列系统
• Kafka:
  Apache下的一个子项目。特点：高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统。适合处理海量数据。

JMS中支持的消息类型:

• TextMessage–一个字符串对象
• MapMessage–一套名称-值对
• ObjectMessage–一个序列化的 Java 对象（可传递集合）
• BytesMessage–一个字节的数据流
• StreamMessage – Java 原始值的数据流

JMS消息传递类型

两种发送模式

• 点对点模式: 一个发送方, 一个接收方. 也可以多个发送方, 一个接收方, 主要是接收方必须是第一个.

• 订阅发布模式: 一个发送方, 多个接收方. 发送方也可以是多个, 主要看接收方, 接收方必须是多个

ActiveMQ小例子

代码只做了解，因为不同的jms，代码完全不同，所以之后会通过spring来进行管理

点对点

当发送方发送数据之后，接收方不论是先启动还是后启动，都能够接收到数据。

接收方

package cn.itcast.queue;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;/*** 点对点模式 接收方*/
public class QueueConsumer {public static void main(String[] args) throws Exception{//1.创建连接工厂ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.200.128:61616");//2.获取连接Connection connection = connectionFactory.createConnection();//3.启动连接connection.start();//4.获取session  (参数1：是否启动事务,参数2：消息确认模式)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//5.创建队列对象Queue queue = session.createQueue("test-queue");//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) {// TODO Auto-generated catch blocke.printStackTrace();}}});//8.等待键盘输入System.in.read();//9.关闭资源consumer.close();session.close();connection.close();}
}

发送方

package cn.itcast.queue;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;/*** 点对点模式 发送方*/
public class QueueProducer {public static void main(String[] args) throws Exception{//1.创建连接工厂ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.200.128:61616");//2.获取连接Connection connection = connectionFactory.createConnection();//3.启动连接connection.start();//4.获取session  (参数1：是否启动事务,参数2：消息确认模式)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//5.创建队列对象，指定发送的队列名称，可以随意起名，但发送到哪里，就要从哪里接收Queue queue = session.createQueue("test-queue");//6.创建消息生产者MessageProducer producer = session.createProducer(queue);//7.创建消息TextMessage textMessage = session.createTextMessage("欢迎来到神奇的品优购世界");//8.发送消息producer.send(textMessage);//9.关闭资源producer.close();session.close();connection.close();}
}

订阅发布

发送方发完数据之后，数据就被发送出去了，如果接收方没有提前启动，则无法收到数据。

发送方

package cn.itcast.topic;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;/*** 订阅发布模式：发送方*/
public class TopicProducer {public static void main(String[] args) throws Exception{//1.创建连接工厂ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.200.128:61616");//2.获取连接Connection connection = connectionFactory.createConnection();//3.启动连接connection.start();//4.获取session  (参数1：是否启动事务,参数2：消息确认模式)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//5.创建主题对象Topic topic = session.createTopic("test-topic");//6.创建消息生产者MessageProducer producer = session.createProducer(topic);//7.创建消息TextMessage textMessage = session.createTextMessage("欢迎来到神奇的品优购世界");//8.发送消息producer.send(textMessage);//9.关闭资源producer.close();session.close();connection.close();}
}

n个接收方

package cn.itcast.topic;import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;/*** 订阅发布模式：接收方*/
public class TopicConsumer1 {public static void main(String[] args) throws Exception{
//1.创建连接工厂ConnectionFactory connectionFactory=new ActiveMQConnectionFactory("tcp://192.168.200.128:61616");//2.获取连接Connection connection = connectionFactory.createConnection();//3.启动连接connection.start();//4.获取session  (参数1：是否启动事务,参数2：消息确认模式)Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);//5.创建主题对象//Queue queue = session.createQueue("test-queue");Topic topic = session.createTopic("test-topic");//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) {// TODO Auto-generated catch blocke.printStackTrace();}}});//8.等待键盘输入System.in.read();//9.关闭资源consumer.close();session.close();connection.close();}
}

如何保证消息服务器不丢消息

答: 我们使用的是activemq消息服务器, 它有两种发送模式点对点和订阅发布

• 点对点:
  不用管, 消息服务器有内部机制保证绝对不会丢数据.
  原理:
  发送方将消息发送给消息服务器后, 消息服务器会将队列中的消息发送给接收方, 接收方接收到消息后会给消息服务器返回一个响应, 消息服务器接收到响应后认为消息接收成功, 将队列中这个发送完的消息删除。
  如果消息服务器在一定时间内没有接受到响应, 则认为消息接收方宕机, 这个时候会将发送完的这个消息进行持久化保存, 如果消息接收方服务器重新启动连接消息服务器, 消息服务器会监听到, 会将之前持久化的消息重新发送给消息服务器.
• 订阅发布:
  订阅发布模式没有实现保证数据安全的机制, 如果消息服务器将消息发送给接收方, 接收方接收到就算接收到了，没接收到就算丢了, 也不进行持久化.