AMQ学习笔记 - 06. 可靠消息传送

2023-01-27 22:36:44

概述

本文介绍JMS中可能发生消息故障的3个隐患阶段，以及确保消息安全的3种保障机制。

故障分析

在介绍可靠传送的确保机制之前，先分析消息在传送的过程中可能在哪个阶段出现问题。

1.两个跃点

跃点的含义在于消息的持有者发生变化，如发送使消息由Producer持有变成JMS Provider持有。在消息传送的过程中，共有2个跃点：
  1. 发送跃点
    Producer将消息发送到JMS Provider的目的地
  2. 接收跃点
    Consumer从JMS Provider的目的地获取消息
AMQ学习笔记 - 06. 可靠消息传送

2.三个隐患阶段

在消息传送过程中，可能在三个不同的阶段出现问题：
  1. 发送阶段
    Producer发送消息到目的地，可能会失败。
  2. 消息缓存阶段
    JMS Provider异常导致目的地中缓存的数据丢失。
  3. 接收阶段
    Consumer从目的地接收消息，可能会失败。
AMQ学习笔记 - 06. 可靠消息传送

可靠传送保障机制

为了解决在隐患阶段可能出现的问题，JMS体系定义了几个保障机制：确认、事务、持久化。

1.Acknowledge - 确认

确认是客户端（Producer或Consumer）和JMS Provider之间，为了确保Message的可靠传送而发送消息进行确认的机制。
这个机制可以用于解决发送阶段、接收阶段的安全隐患。

1.1.对于Producer

确认的机制是：生成方调用的send()方法会被阻塞，直到JMS Provider收到了Message、存储到目的地，并发送确认给生成方，阻塞才被解除。
所以，确认机制对于生成方来说是透明的。

1.2.对于Consumer

确认的机制是：消费者从目的地接收Message之后，发送接收确认给JMS Provider，JMS Provider收到确认后，将从目的地中删除该消息，发送处理确认给消费者。
对于消费者而言，有几种确认机制：
  • AUTO_ACKNOWLEDGE
    自动确认：会话对收到的每条Message，自动发送接收确认；然后会话线程阻塞，直到收到了JMS Provider发来的处理确认。
    接收确认的发送时机：同步接收中，调用receive()方法成功返回；异步接收中，MessageListener的onMessage(Message)方法被调用，并成功返回。
  • CLIENT_ACKNOWLEDGE
    客户端确认：客户端调用Message#acknowledge()发送接收确认[1]；然后会话线程阻塞，直到收到了JMS Provider发来的处理确认。
    接收确认的发送时机：显示调用Message#acknowledge()方法。
    注：
    [1] 每次确认不是只对当前的Message进行确认，而是对自上次确认以来的所有Message进行确认.
  • DUPS_OK_ACKNOWLEDGE
    消息可重复确认：这个机制行为上表现的和AUTO_ACKNOWLEDGE一样，具有延迟确认的特点。这个机制可能会导致收到重复的消息。
    接收确认的发送时机：测试中，PTP Mode每一条消息都会即时确认；Pub/Sub Mode在接收的消息数，每超过prefetch size阀值[1]一半的时候确认一次。
    注：
    [1] 在brokerURL中可以指定参数jms.prefetchPolicy.topicPrefetch作为prefetch size阀值。你可能会猜测：应该也有参数jms.prefetchPolicy.queuePrefetch，可以让DUPS_OK_ACKNOWLEDGE在PTP Mode中也批量确认。但是实验的结果是每一条消息都会即时确认，就好像DUPS_OK_ACKNOWLEDGE只是针对Pub/Sub设定的。

1.3.确认机制的设置

javax.jms.Session针对三种确认机制，分别定义了整型常量：
  • AUTO_ACKNOWLEDGE - 自动确认
  • CLIENT_ACKNOWLEDGE - 客户端确认
  • DUPS_OK_ACKNOWLEDGE - 可重复确认
 
javax.jms.Connection提供了设置的方法：
  • createSession(boolean transacted, int acknowledgeMode):Session
    transacted - 设置事务，后面才会谈到事务，这里设置false就好了
    acknowledgeMode - 设置确认机制
 
确认机制的设置是对于有接收需要的Session来讲的，如果客户端是纯粹的Producer，确认机制的设置应该是被忽略的。鉴于JMS API只提供了上面一个方法来获取Session实例，建议在Producer-Client端获取Session时，使用Connection#createSession(false, Session.AUTO_ACKNOWLEDGE)。对于纯粹的消费者，或者同时作为生产者和消费者，要针对消费的需要来考虑选择哪一个确认机制。

1.4.确认机制的获取

javax.jms.Session提供获取的方法：
  • getAcknowledgeMode():int
    返回结果有4种，除了上述的3种确认机制之外，还有一个Session.SESSION_TRANSACTED表示Session启用了事务。

1.5.测试

关于确认机制的测试，参考确认机制的测试。

2.Transaction - 事务

事务用于将多个操作组成一个原子操作，要么全部成功，要么全部失败。与事务相关的两个指令：提交、回滚。当确认所有的操作都是正确的时候，执行提交，这些操作就会生效；当其中有一个操作失败的时候，执行回滚，所有已执行的操作就会撤销。
事务的原理：类比于数据库中的事务，原理相同。事务都是用于对于多个修改性质的操作进行绑定，以这些个修改操作要么全部成功，要么全部失败的保障，来确保数据的一致性。数据库中的事务是对修改数据的行为进行绑定，JMS中的事务是对修改目的地中消息的行为（发送消息、接收消息）进行绑定。
 
如果你对JDBC中的事务有所了解，对比其和JMS中的事务，兴许能帮助你记忆JMS中的事务：
对比内容\事务 JDBC事务 JMS事务
范围 java.sql.Connection实例的生命周期之内 javax.jms.Session实例的生命周期之内
设置和获取 java.sql.Connection:
  • setAutoCommit(false)
    设置事务
  • getAutoCommit():boolean
    获取事务
 javax.jms.Connection:
  • createSession(true, *)
    设置事务。
    第二个参数可以设置确认模式，可以是任意值：启用了事务就会忽略确认模式。
    不过建议使用Session.SESSION_TRANSACTED

javax.jms.Session:

  • getTransacted():boolean
    获取事务
操作 java.sql.Connection:
  • commit():void
    提交
  • rollback():void
    回滚
 javax.jms.Session:
  • commit():void
    提交
  • rollback():void
    回滚
 
事务的设置和确认机制的设置使用同一个方法，那么有必要把确认机制和事务进行对比：
对比内容\机制 确认机制 事务机制
本质 确保消费者能够获取到消息，在收到消费者的确认之后，才会将消息从目的地移除。 将多个操作绑在一起，成为原子操作，从而有相同的操作结果（成功或失败）。
客户端编程 在Producer端是透明的，只能在Consumer端进行控制。 可以在Producer和Consumer端实施控制。
功能性 确认消息已收到。 事务的commit，提供类似于批量确认的功能。
不要单纯为了批量确认而使用事务，要从业务的需要上考虑使用事务。
因为事务机制提供了确认的功能，所以确认机制和事务机制是互斥的。Connection#createSession(boolean transacted, int acknowledgeMode)方法用于设置事务或确认：
  • 如果第一个参数为true，表示启用事务，第二个参数就会被忽略（建议传入Session.SESSION_TRANSACTED）
  • 如果第一个参数为false，表示不启用事务，第二个参数用于设置确认模式，就有效了。

2.1.使用模型

 try {

     // ...

     javax.jms.Connection connection = ...;

     javax.jms.Session session = connection.createSession(true, Session.SESSION_TRANSACTED);

     // send or receive operates

     session.commit(); // commit at lase

 } catch(JMSException e) {

     session.rollback(); // rollback when exception happens

 } finally {

     // close resources

 }

2.2.测试

关于事务的测试，参考事务的测试。

3.持久化模式

确认模式和事务模式足以[1]处理发送阶段、接收阶段的安全隐患，持久化的机制则用来处理消息在目的地阶段的安全隐患。
如果Producer选择不使用持久化，则消息缓存在内存中，虽然可以获得高吞吐率，但是一旦JMS Provider宕掉，就会导致消息的丢失。非持久化具有高吞吐量和低可靠性的特点。
如果Producer选择持久化，则JMS Provider会将消息存到物理媒介上（文件、数据库），Consumer获取消息，也是从物理媒介读取，吞吐量受到影响，但是即使JMS Provider宕掉，消息也不会丢失。持久化具有低吞吐量和高可靠性的特点。
针对持久化，JMS Provider可以提供多种持久化方案，比如持久化到本地文件、不同的数据库等，不过这些不是Producer需要关心的，Producer只需要告诉JMS Provider要不要持久化消息就好了。
持久化方案的选择，在%ActiveMQ_HOME%\conf\activemq.xml文件，<persistenceAdapter />元素下配置。可以参考ActiveMQ持久化，这里不介绍这块内容。
 
补充：
[1] 实际上，在发送阶段需要我们自己提供容错方案 - 比如对发送失败的消息，缓存到本地文件等。

3.1.持久化设置

javax.jms.MessageProducer:
  • setDeliveryMode(int):void
    - 设置此producer实例的默认递送模式
  • send(Message message):void
    - 以默认的deliveryMode发送消息
  • send(Message message, int deliveryMode, int priority, long timeToLive):void
    - 以指定的deliveryMode发送消息
 
javax.jms.DeliveryMode
  • static int NON_PERSISTENT
    非持久化模式
  • static int PERSISTENT
    持久化模式
 
设置递送模式的代码：
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // 非持久化

producer.setDelicertMode(DeliveryMode.PERSISTENT); // 持久化

3.2.测试

参考

  1. 第 2 章 客户端编程模型
    本文的骨架是参考这篇文章整理的
  2. ActiveMQ讯息传送机制以及ACK机制
    这篇文章对ACK的机制讲的比较细，我尝试了optimizeACK + prefetch + AUTO_ACKNOWLEDGE的预取机制，以及DUPS_OK_ACKNOWLEDGE在Pub/Sub模式下的延迟确认机制。
  3. ActiveMQ持久化
    介绍了几种常见的持久化方案。
上一篇：springboot整合redis(注解形式)


下一篇：http 返回码 405 解决方案之一