• 欢迎访问 winrains 的个人网站!
  • 本网站主要从互联网整理和收集了与Java、网络安全、Linux等技术相关的文章,供学习和研究使用。如有侵权,请留言告知,谢谢!

从JMS规范了解ActiveMQ

消息队列 winrains 来源:匠丶 1年前 (2019-08-31) 33次浏览

ActiveMQ 简介

ActiveMQ 是完全基于 JMS 规范实现的一个消息中间件产品。是 Apache 开源基金会研发的消息中间件。ActiveMQ主要应用在分布式系统架构中,帮助构建高可用、高性能、可伸缩的企业级面向消息服务的系统。
ActiveMQ 特性:
1、多语言和协议编写客户端,语言:java/C/C++/C#/Ruby/Perl/Python/PHP,应用协议:openwire/stomp/REST/ws/notification/XMPP/AMQP;
2、完全支持 jms1.1 和 J2ee1.4 规范;
3、对 spring 的支持,ActiveMQ 可以很容易内嵌到 spring模块中。

从 JMS 规范来了解 ActiveMQ

JMS 定义,Java 消息服务(Java Message Service)是 java 平台中关于面向消息中间件的 API,用于在两个应用程序之间,或者分布式系统中发送消息,进行异步通信。
JMS 规范的目的是为了使得 Java 应用程序能够访问现有 MOM (消息中间件)系统,形成一套统一的标准规范,解决不同消息中间件之间的协作问题。
JMS的体系结构,如下图所示:

生产者发送消息到ActiveMQ:

 ConnectionFactory connectionFactory= new ActiveMQConnectionFactory("tcp://192.168.11.153:61616");
 Connection connection=connectionFactory.createConnection();
 connection.start();
 Session session=connection.createSession (Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);
 //创建目的地
 Destination destination=session.createQueue("myQueue");
//创建发送者
 MessageProducer producer=session.createProducer(destination);
 producer.setDeliveryMode(DeliveryMode.PERSISTENT);
 TextMessage message = session.createTextMessage("Hello World:"+i);
//Text   Map  Bytes  Stream  Object
 producer.send(message);

消费者消费消息:

 ConnectionFactory connectionFactory= new ActiveMQConnectionFactory("tcp://192.168.11.153:61616");
 Connection connection=connectionFactory.createConnection();
 connection.start();
 Session session=connection.createSession (Boolean.FALSE,Session.AUTO_ACKNOWLEDGE);
 //创建目的地
 Destination destination=session.createQueue("myQueue");
//创建消费者
 MessageConsumer consumer=session.createConsumer(destination);
 TextMessage message = (TextMessage) consumer.receive();

JMS 的基本功能

消息传递域:
JMS 规范中定义了两种消息传递域:点对点(point-to-point ) 消 息 传 递 域 和 发 布 / 订 阅 消 息 传 递 域(publish/subscribe)。
点对点消息传递域:

  1. 每个消息只能有一个消费者;
  2. 消息的生产者和消费者之间没有时间上的相关性。无论消费者在生产者发送消息的时候是否处于运行状态,都可以提取消息。

发布订阅消息传递域:

  1. 每个消息可以有多个消费者;
  2. 生产者和消费者之间有时间上的相关性。订阅一个主题的消费者只能消费自它订阅之后发布的消息。JMS 规范允许客户创建持久订阅,这在一定程度上降低了时间上的相关性要求。持久订阅允许消费者消费它在未处于激活状态时发送的消息。

消息组成结构:
JMS 消息组成部分:消息头、属性、消息体。
消息头:
消息头(Header) – 消息头包含消息的识别信息和路由信息,消息头包含一些标准的属性如:
JMSDestination 消息发送的目的地(queue或者topic);
JMSDeliveryMode 传送模式。持久模式和非持久模式;
JMSPriority 消息优先级(优先级分为 10 个级别,从 0(最低)到 9(最高),如果不设定优先级,默认级别是4;
JMSMessageID 唯一识别每个消息的标识。
属性:
按类型可以分为应用设置的属性,标准属性和消息中间件定义的属性。

  1. 应用程序设置和添加的属性,比如Message.setStringProperty(“key”,”value”);
  2. JMS 定义的属性,使用“JMSX”作为属性名的前缀;
  3. JMS provider 特定的属性。

消息体:
就是我们需要传递的消息内容,JMS API 定义了 5 中消息体格式,可以使用不同形式发送接收数据,并可以兼容现有的消息格式,其中包括:

格式
TextMessage java.lang.String 对象,如 xml 文件内容
MapMessage key是 String 对象,value类型可以是 Java 任何基本类型
BytesMessage 字节流
StreamMessage Java 中的输入输出流
ObjectMessage Java 中的可序列化对象
Message 没有消息体,只有消息头和属性

持久订阅:

  1. 持久订阅者和非持久订阅者针对的是 Pub/Sub
  2. 当 Broker 发送消息给订阅者时,如果订阅者处于未激活状态状态:持久订阅者可以收到消息,而非持久订阅者则收不到消息。

这种方式也有一定的影响:当持久订阅者处于未激活状态时,Broker 需要为持久订阅者保存消息;如果持久订阅者订阅的消息太多则会溢出。

JMS 消息的可靠性机制

理论上来说,我们需要保证消息中间件上的消息,只有被消费者确认过以后才会被签收,相当于我们寄一个快递出去,收件人没有收到快递,就认为这个包裹还是属于待签收状态,这样才能保证包裹能够安全达到收件人手里。消息中间件也是一样。
消息的消费通常包含 3 个阶段:客户接收消息、客户处理消息、消息被确认。
首先,简单了解 JMS 的事务性会话和非事务性会话的概念。
事务型会话:
在事务状态下进行发送操作,消息并未真正投递到中间件,而只有进行 session.commit 操作之后,消息才会发送到中间件,再转发到适当的消费者进行处理。如果是调用rollback 操作,则表明,当前事务期间内所发送的消息都取消。通过在创建 session 的时候使用 true or false 来决定当前的会话是事务性还是非事务性。

connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);

在事务性会话中,消息的确认是自动进行,也就是通过session.commit()以后,消息会自动确认。并且必须保证发送端和接收端都是事务性会话。
非事务型会话:
消 息 何 时 被 确 认 取 决 于 创 建 会 话 时 的 应 答 模 式(acknowledgement mode),有三个可选项:
1、Session.AUTO_ACKNOWLEDGE,当客户成功的从 receive 方法返回的时候,或者从MessageListenner.onMessage 方法成功返回的时候,会话自动确认客户收到消息。
2、Session.CLIENT_ACKNOWLEDGE,客户通过调用消息的 acknowledge 方法确认消息。
3、Session.DUPS_ACKNOWLEDGE,消息延迟确认。指定消息提供者在消息接收者没有确认发送时重新发送消息,这种模式不在乎接受者收到重复的消息。
消息的持久化
消息的持久化存储也是保证可靠性最重要的机制之一,也就是消息发送到 Broker 上以后,如果 broker 出现故障宕机了,那么存储在 broker 上的消息不会丢失。
对于非持久的消息,JMS provider 不会将它存到文件/数据库等稳定的存储介质中。也就是说非持久消息驻留在内存中,如果 jms provider 宕机,那么内存中的非持久消息会丢失。
对于持久消息,消息提供者会使用存储-转发机制,先将消息存储到稳定介质中,等消息发送成功后再删除。

消息的幂等

为什么会有重复消息
主要是分布式系统可能出现网络不稳定、应用宕机等异常情况,简单分为:
1、生产者发送重复消息,当一条消息成功发送到broker并持久化到硬盘之后,此时出现了网络抖动、客户端宕机、应用重启,导致broker回复客户端失败,客户端没有收到发送成功的通知,会重试三次发送给broker,这时消费者可能收到两条内容相同、消息id也相同的消息;重要业务系统通常会针对发送失败的消息定时重发,这时消费组可能收到两条内容相同、消息id不同的消息;
2、broker发送重复消息,当消费者成功执行消费业务逻辑之后,此时客户端与broker之间出现网络抖动、客户端宕机、应用重启等意外情况,提交给broker的消费进度更新失败。为了确保消息至少被消费一次,此时broker会再次投递,消费者便收到了两条消息内容相同的消息。
如何去重
1、在消费业务逻辑之前,去除业务唯一键,如订单id、任务id等,判断是否在Mysql或Redis存在,存在则跳过消费,对于重要消息,如订单、交易等需要考虑消费的原子性。禁止使用msgid来去重,由于网络抖动或者业务补偿可能出现msgid不同但消息内容的情况。
2、使用业务层面的状态机去重

作者:匠丶

来源:https://www.jianshu.com/p/136ef95765d4


版权声明:文末如注明作者和来源,则表示本文系转载,版权为原作者所有 | 本文如有侵权,请及时联系,承诺在收到消息后第一时间删除 | 如转载本文,请注明原文链接。
喜欢 (1)