【ActiveMQ】之 ActiveMQ 的高级特性

港控/mmm° 2023-02-28 11:14 2阅读 0赞

## 异步投递 ##

--------------------

ActiveMQ 支持同步和异步两种发送的模式将消息发送到 broker，模式的选择对发送延时有巨大的影响。producer 能达到的产出率（**产出率=发送数据总量/时间**）主要受发送延时的影响，使用异步发送可以显著的提高发送性能。

对于一个慢消费者（**数据投送快但数据消费满**）来说，使用同步发送消息可能出现 Producer 堵塞以及 broker 消息数量积压的情况，所以慢消费者适合使用异步发送。

**ActiveMQ 默认使用异步发送模式**，除非明确指定使用同步发送的方式或者在未使用事务的前提下发送持久化的消息。

如果没有使用事务但发送的却又是持久化的消息，每一次发送都是同步发送且会阻塞 producer 知道 broker 返回一个确认，表示消息已经被安全的持久化到磁盘。确认机制提供了消息安全的保障，但同时会阻塞客户端从而带来了很大的延时。

（**在高性能的应用中，如果允许在失败的情况下有少量的数据丢失，则可以使用异步发送来提高上产率**）

通常在发送消息比较密集的情况下都会使用异步发送来最大化 producer 端的发送效率，从而提高 producer 的性能。与此同时也会带来其它额外的问题：

*  需要消耗较多的 Client 端的内存同时也会导致 broker端性能消耗增加，特别是在慢消费者的情况下。
 *  它不能有效的确保消息的发送成功，使用 useAsyncSend=true 的情况下，客户端需要容忍消息丢失的可能。

**开启异步投递的方式**

配置 URL：

cf = new ActiveMQConnectionFactory("tcp://locahost:61616?jms.useAsyncSend=true");

配置 ConnectionFactory（记住，对象必须是 ActiveMQConnectionFactory）：

((ActiveMQConnectionFactory)connectionFactory).setUseAsyncSend(true);

配置 Connection （记住，对象必须是 ActiveMQConnection）

((ActiveMQConnection)connection).setUseAsyncSend(true);

**如何确保异步投递发送成功**

异步发送丢失消息的场景是：生产者设置 useAsyncSend=true，使用 producer.send(msg) 持续发送消息，由于消息不阻塞，生产者会认为所有 send 的消息均被成功发送至 MQ。如果此时 MQ 突然宕机，那么生产者内存中尚未被发送至 MQ 的消息都会丢失。

正确的异步发送方法是**需要接受回调**。

同步发送和异步发送的区别就在这里，同步发送等 send 不阻塞了就表示一定发送成功了，异步发送需要接受回执并有客户端再判断一次是否发送成功。

异步发送设置回调代码实例如下：

public class JmsProducerAsync { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String QUEUE_NAME = "my-queue";
    
        @Test
        public void producer() throws JMSException { 
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            activeMQConnectionFactory.setUseAsyncSend(true);    // 设置异步发送
    
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue(QUEUE_NAME);
    
            // 注意，需要使用的是 ActiveMQMessageProducer
            ActiveMQMessageProducer activeMQMessageProducer = (ActiveMQMessageProducer) session.createProducer(queue);
            activeMQMessageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
    
            TextMessage textMessage = null;
            for (int i = 0; i < 6; i++) { 
                textMessage = session.createTextMessage("message " + i);
    
                // 使用 ID 标识消息
                textMessage.setJMSMessageID("message id : " + UUID.randomUUID().toString());
                String messageID = textMessage.getJMSMessageID();
    
                // 设置异步发送回调
                activeMQMessageProducer.send(textMessage, new AsyncCallback() { 
                    @Override
                    public void onSuccess() { 
                        System.out.println("异步消息发送成功 " + messageID);
                    }
    
                    @Override
                    public void onException(JMSException exception) { 
                        // 异步发送失败后的处理
                        System.out.println("异步消息发送失败 " + messageID);
                    }
                });
    
            }
    
            activeMQMessageProducer.close();
            session.close();
            connection.close();
    
            System.out.println("======== 消息发布到 MQ 完成 ==========");
        }
    }

更多配置参考官网介绍：[http://activemq.apache.org/async-sends][http_activemq.apache.org_async-sends]

## 消息的延时投递和定时投递 ##

--------------------

根据官网：[http://activemq.apache.org/delay-and-schedule-message-delivery][http_activemq.apache.org_delay-and-schedule-message-delivery] 的介绍，要开启延时投递和定时投递，需要在配置文件 activemq.xml 文件中的 broker 标签中添加 **schedulerSupport** 属性：

时投递和定时投递的四大属性：

AMQ_SCHEDULED_DELAY		long 	// 延时投递的延时时间
    AMQ_SCHEDULED_PERIOD	long	// 定时投递的时间间隔
    AMQ_SCHEDULED_REPEAT	int		// 投递的重复次数
    MQ_SCHEDULED_CRON		String	// Corn表达式

具体 Producer 代码实例如下：

public class JmsProducerSchedule { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String QUEUE_NAME = "my-queue";
    
        public static void main(String[] args)  throws JMSException { 
    
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
    
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Queue queue = session.createQueue(QUEUE_NAME);
            MessageProducer messageProducer = session.createProducer(queue);
    
            // 设置时间
            long delay = 3 * 1000;  // 延时3秒
            long period = 4 * 1000; // 定时4秒
            int repeat = 5;         // 重复5次
    
            for (int i = 0; i < 3; i++) { 
                TextMessage message = session.createTextMessage("schedule message " + i);
    
                // 设置消息 Schedule 参数
                message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay);
                message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, period);
                message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, repeat);
    
                messageProducer.send(message);
            }
    
            messageProducer.close();
            session.close();
            connection.close();
    
            System.out.println("======== 消息发布到 MQ 完成 ==========");
        }
    }

设置成功后，哪怕 producer 程序运行完毕退出，消费者那边也会收到定时产生的消息。

## 消息重试机制 ##

--------------------

**引起消息重发的情况：**

1.  客户端开启了事务并在 session 中调用了 rollback()
2.  客户端开启了事务但没有调用 commit() 或者调用 commit() 之前就关闭了
3.  客户端在自动签收模式（CLIENT\_ACKNOWLEDGE）下，在 session 中调用了重试 recover()

**默认的重发时间间隔是 1s，默认的重发次数为 6 次。**

**有毒消息（Poison Ack）:**

一个消息被重新投递（redelivedred）超过默认的最大重发次数（默认6次）时，消费端会给 MQ 发送一个有毒标志 “poison ack”，表示这个消息有毒，告诉 broker 不要再发了。这个时候 broker 就会把这个消息放到死信队列（DLQ）中。

修改消费重试机制的参数：

具体参数类型和修改方式可以参考官网：[http://activemq.apache.org/redelivery-policy][http_activemq.apache.org_redelivery-policy]，下面我们以修改重发间隔时间和最大重发次数为例，具体修改代码如下：

public class JmsConsumerRedelivery { 
    
        public static final String ACTIVEMQ_URL = "tcp://localhost:61616";
        public static final String QUEUE_NAME = "my-queue";
    
        public static void main(String[] args) throws JMSException, IOException { 
    
            // 创建连接工厂，按照指定的URL地址，采用默认用户名和密码
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
    
            // 创建消息重投策略
            RedeliveryPolicy policy = new RedeliveryPolicy();
            policy.setRedeliveryDelay(1500);    // 重投时间间隔
            policy.setMaximumRedeliveries(3);   // 重投次数
            activeMQConnectionFactory.setRedeliveryPolicy(policy);
    
            // 连接连接工厂获得 connect 连接池并启动访问
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
    
            // 创建会话 session 并开启事务
            Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
    
            // 创建队列目的地
            Queue queue = session.createQueue(QUEUE_NAME);
    
            // 创建消息的消费者
            MessageConsumer messageConsumer = session.createConsumer(queue);
    
            // 从 MQ 队列中获取消息
            while (true){ 
                TextMessage textMessage = (TextMessage) messageConsumer.receive(2000L);
                if (textMessage != null) { 
                    System.out.println("消费者接收到的消息 ====> " + textMessage.getText());
                } else { 
                    break;
                }
            }
            // 不提交事务导致重复消费
            // session.commit();
    
            // 7、关闭资源
            messageConsumer.close();
            session.close();
            connection.close();
        }
    }

## 死信队列 ##

--------------------

更多有关死信队列的信息可以参考官网：[http://activemq.apache.org/message-redelivery-and-dlq-handling][http_activemq.apache.org_message-redelivery-and-dlq-handling]

正如前面所介绍，ActiveMQ 中引入了 \*\*死信队列（Dead Letter Queue）\*\*的概念，即：一条消息再次重新发送次数达到最大次数（默认6次）时，就会把该消息移动到死信队列中。有了死信队列，开发人员就可以在这个 Queue中查看出错的消息，要么自动处理要么进行人工干预处理。

**死信队列的配置**

**1、共享死信队列策略（SharedDeadLetterStrategy）**

ActiveMQ broker 的默认策略是将所有的 DeadLetter 保存在一个共享的队列中，共享队列的默认队列名为：ActiveMQ.DLQ，可以通过修改 activemq.xml 配置文件来修改：

**2、独立死信队列策略（IndividualDeadLetterStrategy ）**

把 DeadLetter 放入各自的死信通道中：

*  对于 Queue 而言，死信通道的前缀默认为：ActiveMQ.DLQ.Queue
 *  对于 Topic 而言，死信通道的前缀默认为：ActiveMQ.DLQ.Topic

比如队列 Order，那么它对应的死信通道为：ActiveMQ.DLQ.Queue.Order，我们使用 **queuePrefix、topicPrefix** 来指定上述前缀。

默认情况下，无论是 Topic 还是 Queue，broker 将使用 Queue 来保存 DeadLetter，即 死信 通道通常为 Queue，不过我们可以指定为 Topic:

属性 useQueueForQueueMessages 的值表示是否将 Topic 的 DeadLetter 保存在 Queue 中，默认为 true，false 表示如果原队列为 Topic 则保存在 Topic 队列中。

（需要留意到是 <policyEntry queue=“Order”> 中的 queue 属性值如果为 ‘>’ 则表示策略应用在所有队列中）

除此之外，我们还是可以使用 processExpired 属性来配置是否将过期消息放入到死信队列中，默认为 true：

还可以使用 processNonPersistent 属性来配置是否把非持久的死消息放到死信队列中，默认是不放的：

## 如何保证消息不重复消费 ##

--------------------

1、幂等性，同样的参数调用同一个接口，调用多少次结果都是一个

2、如果消息是做数据库的插入操作，可以给这个消息做一个唯一主键，这时如果出现重复消费的情况，就会导致主键冲突，从而可以避免数据库出现脏读数据。

3、使用第三服务方来做消费记录，比如 redis，给消息分配一个全局 ID，只要消费该消息，将 <id, message> 以 K-V 形式写入 redis，在消费者开始消费前，先去 redis 中查询有没有消费记录即可。

[http_activemq.apache.org_async-sends]: http://activemq.apache.org/async-sends
[http_activemq.apache.org_delay-and-schedule-message-delivery]: http://activemq.apache.org/delay-and-schedule-message-delivery
[http_activemq.apache.org_redelivery-policy]: http://activemq.apache.org/redelivery-policy
[http_activemq.apache.org_message-redelivery-and-dlq-handling]: http://activemq.apache.org/message-redelivery-and-dlq-handling