登录 注册

Active MQ 高级特性和用法（二）：消息的可靠性、通配符式订阅与死信队列

本贴最后更新于 1720 天前，其中的信息可能已经水流花落

Active MQ 高级特性和用法（二）：消息的可靠性、通配符式订阅与死信队列

一、消息的可靠性

消息发送成功后，接收端接收到了消息。然后进行处理，但是可能由于某种原因，高并发也好，IO 阻塞也好，反正这条消息在接收端处理失败了。而点对点的特性是一条消息，只会被一个接收端给接收，只要接收端 A 接收成功了，接收端 B，就不可能接收到这条消息，如果是一些普通的消息还好，但是如果是一些很重要的消息，比如说用户的支付订单，用户的退款，这些与金钱相关的，是必须保证成功的，那么这个时候要怎么处理呢？
必须要保证消息的可靠性，除了消息的持久化，还包括两个方面，一是生产者发送的消息可以被 ActiveMQ 收到，二是消费者确实收到了 ActiveMQ 发送的消息

生产者端的可靠性

1、send()方法

在生产者端，我们会使用 send() 方法向 ActiveMQ 发送消息，默认情况下，持久化消息以同步方式发送，send() 方法会被阻塞，直到 broker 发送一个确认消息给生产者，这个确认消息表示 broker 已经成功接收到消息，并且持久化消息已经把消息保存到二级存储中。

2、测试 send()方法

//循环发送消息
for (int i = 0; i < SENDNUM; i++) {
    String msg = "发送消息" + i + " " + System.currentTimeMillis();
    TextMessage textMessage = session.createTextMessage(msg);
    System.out.println("标准用法:" + msg);
    messageProducer.send(textMessage);
}

    我们在 send 方法上打一个断点，可以看到 send 方法每执行一次，ActiveMQ 管理控制台增加一条入队消息，数据库中增加一条消息。

3、事务消息

事务中消息（无论是否持久化），会进行异步发送，send() 方法不会被阻塞。但是 commit 方法会被阻塞，直到收到确认消息，表示 broker 已经成功接收到消息，并且持久化消息已经把消息保存到二级存储中。

4、测试事务消息

//循环发送消息
for (int i = 0; i < SENDNUM; i++) {
    String msg = "发送消息" + i + " " + System.currentTimeMillis();
    TextMessage textMessage = session.createTextMessage(msg);
    System.out.println("标准用法:" + msg);
    messageProducer.send(textMessage);
}
session.commit();

    我们在 session.commit()打一个断点，可以看到 send 方法每执行一次，ActiveMQ 管理控制台和数据库中没有任何变化，只有执行完 session.commit()后 ActiveMQ 管理控制台和数据库中才增加。

5、生产者端可靠性总结

    非持久化又不在事务中的消息，可能会有消息的丢失。为保证消息可以被 ActiveMQ 收到，我们应该采用事务消息或持久化消息。

消费者端的可靠性

ACK_MODE 描述了 Consumer 与 broker 确认消息的方式(时机),比如当消息被 Consumer 接收之后,Consumer 将在何时确认消息。所以 ack_mode 描述的不是 producer 于 broker 之间的关系，而是 customer 于 broker 之间的关系。
对于 broker 而言，只有接收到 ACK 指令,才会认为消息被正确的接收或者处理成功了,通过 ACK，可以在 consumer 与 Broker 之间建立一种简单的“担保”机制.

对消息的确认有 4 种机制(customer 对 broker 进行消息确认)

  1. AUTO_ACKNOWLEDGE = 1 自动确认
  2. LIENT_ACKNOWLEDGE = 2 客户端手动确认
  3. DUPS_OK_ACKNOWLEDGE = 3 自动批量确认
  4. ESSION_TRANSACTED = 0 事务提交并确认

    在创建 Session 时设置消息的确认机制：
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
第一个参数:是否支持事务，如果为 true，则会忽略第二个参数，自动被 jms 服务器设置为 SESSION_TRANSACTED

1、AUTO_ACKNOWLEDGE

  • 客户端自动确认机制。
  • “同步”(receive)方法返回 message 给消息时会立即确认。
  • 在"异步"(messageListener)方式中,将会首先调用 listener.onMessage(message)，如果 onMessage 方法正常结束,消息将会正常确认。如果 onMessage 方法异常，将导致消费者要求 ActiveMQ 重发消息。此外需要注意，消息的重发次数是有限制的，每条消息中都会包含“redeliveryCounter”计数器，用来表示此消息已经被重发的次数，如果重发次数达到阀值，将导致 broker 端认为此消息无法消费,此消息将会被删除或者迁移到"dead letter"通道中。
  • 因此当我们使用 messageListener 方式消费消息时，可以在 onMessage 方法中使用 try-catch,这样可以在处理消息出错时记录一些信息，而不是让 consumer 不断去重发消息；如果你没有使用 try-catch,就有可能会因为异常而导致消息重复接收的问题,需要注意 onMessage 方法中逻辑是否能够兼容对重复消息的判断。

1、LIENT_ACKNOWLEDGE

  • 客户端手动确认，这就意味着 AcitveMQ 将不会“自作主张”的为你 ACK 任何消息，开发者需要自己择机确认。可以用方法： message.acknowledge()，或 session.acknowledge()；效果一样。
  • 如果忘记调用 acknowledge 方法，将会导致当 consumer 重启后，会接受到重复消息，因为对于 broker 而言，那些尚未真正 ACK 的消息被视为“未消费”。
  • 我们可以在当前消息处理成功之后，立即调用 message.acknowledge()方法来"逐个"确认消息，这样可以尽可能的减少因网络故障而导致消息重发的个数；当然也可以处理多条消息之后，间歇性的调用 acknowledge 方法来一次确认多条消息，减少 ack 的次数来提升 consumer 的效率，不过需要自行权衡。

1、DUPS_OK_ACKNOWLEDGE

  • 类似于 AUTO_ACK 确认机制，为自动批量确认而生，而且具有“延迟”确认的特点，ActiveMQ 会根据内部算法，在收到一定数量的消息自动进行确认。在此模式下，可能会出现重复消息，什么时候？当 consumer 故障重启后，那些尚未 ACK 的消息会重新发送过来。

1、ESSION_TRANSACTED

  • 当 session 使用事务时，就是使用此模式。当决定事务中的消息可以确认时，必须调用 session.commit()方法，commit 方法将会导致当前 session 的事务中所有消息立即被确认。在事务开始之后的任何时机调用 rollback()，意味着当前事务的结束，事务中所有的消息都将被重发。当然在 commit 之前抛出异常，也会导致事务的 rollback。

二、通配符式分层订阅

Wildcards 用来支持联合的名字分层体系（federated name hierarchies）。它不是 JMS 规范的一部分，而是 ActiveMQ 的扩展。

ActiveMQ 支持以下三种 wildcards：

  • "." 用于作为路径上名字间的分隔符。
  • "*" 用于匹配路径上的任何名字。
  • ">" 用于递归地匹配任何以这个名字开始的 destination。

    订阅者可以明确地指定 destination 的名字来订阅消息，或者它也可以使用 wildcards 来定义一个分层的模式来匹配它希望订阅的 destination。

通配符测试

    接下来，我们创建了三个 topic 模式的消费者，并启动。

//消费者01,匹配kk.开头的任何destination
destination = session.createTopic("kk.>");
//消费者02，匹配kk.vip.开头的任何destination
destination = session.createTopic("kk.vip.>");
//消费者03，匹配kk.vip.*.redis.cache
destination = session.createTopic("kk.vip.*.redis.cache");

    然后定义三个 topic 模式的生产者，来测试一下：

//生产者01，符合消费者1、2的匹配规则
destination = session.createTopic("kk.vip.program.thread");
//生产者02，符合消费者1、2、3的匹配规则
destination = session.createTopic("kk.vip.program.redis.cache");
//生产者03，符合消费者1的匹配规则
destination = session.createTopic("kk.public.arct.redis.cache");

    测试结果
activemq01.png

三、死信队列 DLQ（Dead Letter Queue）

死信队列是用来保存处理失败或者过期的消息的一种特殊的普通队列。

    当一个消息被重发超过最大重发次数（缺省为 6 次，消费者端可以修改）时，会给 broker 发送一个"有毒标记“，这个消息被认为是有问题，这时 broker 将这个消息发送到死信队列，以便后续处理。

    消息在什么情况下会被重发？

  1. 事务会话被回滚。
  2. 事务会话在提交之前关闭。
  3. 会话使用 CLIENT_ACKNOWLEDGE 模式，并且 Session.recover()被调用。
  4. 自动应答失败

在配置文件(activemq.xml)来调整死信发送策略：
DLQ.png

<policyEntry queue=">"> 
   <deadLetterStrategy> 
	<!--queuePrefix:设置死信队列前缀 -->
	<!--useQueueForQueueMessages:设置使用队列保存死信 -->
	<!--可以设置useQueueForQueueMessages，使用Topic来保存死信 -->
    <individualDeadLetterStrategy queuePrefix="DLQ." useQueueForQueueMessages="true" processExpired="false"/> 
     <!--是否丢弃过期消息-->
     <!--<sharedDeadLetterStrategy processExpired="false" />-->
    </deadLetterStrategy> 
</policyEntry>

制造死信队列

    测试死信队列发送端：

public class DlqProducer {

    //默认连接用户名
    private static final String USERNAME
            = ActiveMQConnection.DEFAULT_USER;
    //默认连接密码
    private static final String PASSWORD
            = ActiveMQConnection.DEFAULT_PASSWORD;
    //默认连接地址
    private static final String BROKEURL
            = ActiveMQConnection.DEFAULT_BROKER_URL;
    //发送的消息数量
    private static final int SENDNUM = 1;

    public static void main(String[] args) {
	    //不是直接使用接口，而是使用Active MQ所提供的工厂，已便于我们进行更多配置信息。
        ActiveMQConnectionFactory connectionFactory;
        ActiveMQConnection connection = null;
        Session session;
        ActiveMQDestination destination;
        MessageProducer messageProducer;

        connectionFactory
                = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEURL);
        try {
            connection = (ActiveMQConnection) connectionFactory.createConnection();
            connection.start();

            session = connection.createSession(true,Session.AUTO_ACKNOWLEDGE);

            destination = (ActiveMQDestination) session.createQueue("TestDlq2");
            messageProducer = session.createProducer(destination);
            for(int i=0;i<SENDNUM;i++){
                String msg = "发送消息"+i+" "+System.currentTimeMillis();
                TextMessage message = session.createTextMessage(msg);

                System.out.println("发送消息:"+msg);
                messageProducer.send(message);
            }
            session.commit();

        } catch (JMSException e) {
            e.printStackTrace();
        }finally {
            if(connection!=null){
                try {
                    connection.close();
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

    测试死信队列的接收端
    通过配置重发策略来制造死信队列

public class DlqConsumer {

    private static final String USERNAME
            = ActiveMQConnection.DEFAULT_USER;//默认连接用户名
    private static final String PASSWORD
            = ActiveMQConnection.DEFAULT_PASSWORD;//默认连接密码
    private static final String BROKEURL
            = ActiveMQConnection.DEFAULT_BROKER_URL;//默认连接地址

    public static void main(String[] args) {
        ActiveMQConnectionFactory connectionFactory;
        ActiveMQConnection connection = null;
        Session session;
        ActiveMQDestination destination;

        MessageConsumer messageConsumer;//消息的消费者

        //实例化连接工厂
        connectionFactory
                = new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEURL);
        //配置策略
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        //限制了重发次数为1
        redeliveryPolicy.setMaximumRedeliveries(1);

        try {
            //通过连接工厂获取连接
            connection = (ActiveMQConnection) connectionFactory.createConnection();
            //启动连接
            connection.start();
            // 拿到消费者端重复策略map
            RedeliveryPolicyMap redeliveryPolicyMap
                    = connection.getRedeliveryPolicyMap();
            //创建session
            session
                    = connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            destination = (ActiveMQDestination) session.createQueue("TestDlq2");
            // 将消费者端重发策略配置给消费者
            redeliveryPolicyMap.put(destination,redeliveryPolicy);
            //创建消息消费者
            messageConsumer = session.createConsumer(destination);
            messageConsumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        System.out.println("Accept msg : "
                                +((TextMessage)message).getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                    throw new RuntimeException("test");
                }
            });
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

消费死信队列

    消费死信队列：

public class ProcessDlqConsumer {

    private static final String USERNAME
            = ActiveMQConnection.DEFAULT_USER;//默认连接用户名
    private static final String PASSWORD
            = ActiveMQConnection.DEFAULT_PASSWORD;//默认连接密码
    private static final String BROKEURL
            = ActiveMQConnection.DEFAULT_BROKER_URL;//默认连接地址

    public static void main(String[] args) {
        ConnectionFactory connectionFactory;//连接工厂
        Connection connection = null;//连接

        Session session;//会话 接受或者发送消息的线程
        Destination destination;//消息的目的地

        MessageConsumer messageConsumer;//消息的消费者

        //实例化连接工厂
        connectionFactory = new ActiveMQConnectionFactory(ProcessDlqConsumer.USERNAME,
                ProcessDlqConsumer.PASSWORD, ProcessDlqConsumer.BROKEURL);

        try {
            //通过连接工厂获取连接
            connection = connectionFactory.createConnection();
            //启动连接
            connection.start();
            //创建session
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //创建一个连接HelloWorld的消息队列
            //destination = session.createTopic("TestDlq");
            destination = session.createQueue("DLQ.>");

            //创建消息消费者
            messageConsumer = session.createConsumer(destination);
            messageConsumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    try {
                        System.out.println("Accept DEAD msg : "
                                +((TextMessage)message).getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}
  • ActiveMQ

    ActiveMQ 是 Apache 旗下的一款开源消息总线系统，它完整实现了 JMS 规范，是一个企业级的消息中间件。

    19 引用 • 13 回帖 • 671 关注
  • 消息队列
    40 引用 • 52 回帖 • 2 关注
烟台 位置
317 6 60 30 54 15 1 15 128

相关帖子

回帖
Active MQ 高级特性和用法（二）：消息的可靠性、通配符式订阅与死信队列

欢迎来到这里！

我们正在构建一个小众社区，大家在这里相互信任，以平等 • 自由 • 奔放的价值观进行分享交流。最终，希望大家能够找到与自己志同道合的伙伴，共同成长。

注册 关于
请输入回帖内容 ...
DongXiaokai0819
烟台
回帖
5
 帖子
23
 积分
489