异步投递及确认成功

同步发送

ActiveMQ官方说异步发送是很多模式下默认的传输方式，但是在发送非事物持久化消息的时候默认使用的是同步发送模式。同步发送时，Producer.send() 方法会被阻塞，直到 broker 发送一个确认消息给生产者，这个确认消息暗示生产者 broker 已经成功地将它发送的消息路由到目标目的并把消息保存到二级存储中。

同步发送持久消息能够提供更好的可靠性，但这潜在地影响了程序的响应速度，因为在接受到 broker 的确认消息之前应用程序或线程会被阻塞。如果应用程序能够容忍一些消息的丢失，那么可以使用异步发送。异步发送不会在受到 broker 的确认之前一直阻塞 Producer.send 方法。

异步发送

使用不同的模式对send 方法的反应时间有巨大的影响，反映时间是衡量ActiveMQ 吞吐量的重要因素，使用异步发送可以提高系统的性能。在默认大多数情况下，AcitveMQ 是以异步模式发送消息。例外的情况：在没有使用事务的情况下，生产者以PERSISTENT 传送模式发送消息。在这种情况下，send 方法都是同步的，并且一直阻塞直到ActiveMQ 发回确认消息：消息已经存储在持久性数据存储中。这种确认机制保证消息不会丢失，但会造成生产者阻塞从而影响反应时间。高性能的程序一般都能容忍在故障情况下丢失少量数据。如果编写这样的程序，可以通过使用异步发送来提高吞吐量（甚至在使用PERSISTENT 传送模式的情况下）。

设置异步发送：

ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("ACTIVEMQ_URL");
activeMQConnectionFactory.setUseAsyncSend(true);

异步发送丢失消息的场景

生产者设置UseAsyncSend=true，使用producer.send(msg)持续发送消息。由于消息不阻塞，生产者会认为所有send的消息均被成功发送至MQ。如果服务端突然宕机，此时生产者端内存中尚未被发送至MQ的消息都会丢失。正确的异步发送方法是需要接收回调的。

producer.send有带上AsyncCallback的方法。该方法中需要重写onSuccess方法和onException方法。onSuccess方法就是表示这条消息成功发送到MQ上，并接收到了MQ持久化后的回调。onException表示MQ返回一个入队异常的回执。在上面的示例中用的是CountDownLatch类在onSuccess中记录。主要是因为onSuccess方法中只能引用final对象。

代码测试：

public class TestMessageProducer {
    private static final String BROKER_URL = "tcp://192.168.81.130:61616";
    //private static final String BROKER_URL = "tcp://192.168.81.130:61616,tcp://192.168.81.130:61617,tcp://192.168.81.130:61618";
    private static final String QUEUE_NAME = "queue-test";
    public static void main(String[] args) throws JMSException {
        ConnectionFactory factory=new ActiveMQConnectionFactory(BROKER_URL);
        ((ActiveMQConnectionFactory) factory).setAlwaysSyncSend(true);//开启异步发送
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(QUEUE_NAME);

        ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(queue);
        System.out.println(producer.getClass().getSimpleName());
        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
        TextMessage message = session.createTextMessage("我是一个消息----helo");
        //        producer.send(message); //因为上面的持久的方式，所以这种试用方式默认是使用同步模式

        producer.send(queue, message, new AsyncCallback() {
            @Override
            public void onSuccess() {
                System.out.println("消费发送成功，所处线程:"+Thread.currentThread().getName());
            }

            @Override
            public void onException(JMSException e) {
                System.out.println("消费发送失败，所处线程:"+Thread.currentThread().getName());
            }
        });
        System.out.println("主程序所处线程为:"+Thread.currentThread().getName());
        producer.close();
        session.close();
        connection.close();
        System.out.println("消息发送成功");
    }
}

延迟投递和定时投递

概述

5.4版的ActiveMQ在ActiveMQ消息代理中内置了一个可选的持久性调度程序。通过在“ Xml配置”中将broker schedulerSupport属性设置为true 可以启用此功能。ActiveMQ客户端可以通过使用以下消息属性来利用延迟传递。

http://activemq.apache.org/delay-and-schedule-message-delivery.html

属性说明

测试案例：延时20秒

TextMessage message = session.createTextMessage("我是一个消息----helo");
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY,20000);
producer.send(message);

测试案例：延时10秒，投递10次，间隔5秒

TextMessage message = session.createTextMessage("我是一个消息----helo");
//多少毫秒后入队
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY,10000);
//隔多少毫秒再投
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD,5000);
//总共投多少次
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT,10);
producer.send(message);

测试案例：使用CRON表达式

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, "0 * * * *");
producer.send(message);

MessageProducer producer = session.createProducer(destination);
TextMessage message = session.createTextMessage("test msg");
message.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_CRON, "0 * * * *");
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, 1000);
message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, 1000);
message.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, 9);
producer.send(message);

其它说明

1、CRON表达式的优先级高于另外三个参数，如果在设置了CRON的同时，也有repeat和period参数，则会在每次CRON执行的时候，重复投递repeat次，每次间隔为period。就是说设置是叠加的效果。例如每小时都会发生消息被投递10次，延迟1秒开始，每次间隔1秒。

2、此处Cron是Unix系统中任务调度器,它使用一个字符串来表示一个任务何时需要被执行。而不是Quartz里边的那个Cron表达式。

消费重试机制

ActiveMQ会在什么情况下重新发送消息？

ActiveMQ中的消息重发，指的是消息可以被broker重新分派给消费者，不一定的之前的消费者。重发消息之后，消费者可以重新消费。消息重发的情况有以下几种。

1、事务会话中，当还未进行session.commit()时，进行session.rollback()，那么所有还没commit的消息都会进行重发。

2、使用客户端手动确认的方式时，还未进行确认并且执行Session.recover()，那么所有还没acknowledge的消息都会进行重发。

3、所有未ack的消息，当进行session.closed()关闭事务，那么所有还没ack的消息broker端都会进行重发，而且是马上重发。

4、消息被消费者拉取之后，超时没有响应ack，消息会被broker重发。

重发指的是消息经过broker重新进行转发给消费者，经过测试，1和2的情况消息重发会发送给原来的消费者，3和4可以转发消息给别的消费者。累计次数超过设置的maximumRedeliveries时消息都会都会进入死信队列。

消息的重发时间间隔和重发次数：间隔 1，次数 6 （6次之后进入死信队列）。

1、collisionAvoidanceFactor：设置防止冲突范围的正负百分比，只有启用useCollisionAvoidance参数时才生效。也就是在延迟时间上再加一个时间波动范围。默认值为0.15。

2、maximumRedeliveries：最大重传次数，达到最大重连次数后抛出异常。为-1时不限制次数，为0时表示不进行重传。默认值为6。

3、maximumRedeliveryDelay：最大传送延迟，只在useExponentialBackOff为true时有效(V5.5) ，假设首次重连间隔为10ms，倍数为2,那么第二次重连时间间隔为20ms，第三次重连时间间隔为40ms，当重连时间间隔大的最大重连时间间隔时，以后每次重连时间间隔都为最大重连时间间隔。默认为-1。

4、initialRedeliveryDelay：初始重发延迟时间，默认1000L。

5、redeliveryDelay：重发延迟时间，当initialRedeliveryDelay=0时生效，默认1000L。

6、useCollisionAvoidance：启用防止冲突功能，默认false。

7、useExponentialBackOff：启用指数倍数递增的方式增加延迟时间，默认false。

8、backoffMultiplier：重连时间间隔递增倍数，只有值大于1和启用useExponentialBackOff参数时才生效。默认是5。

Java配置

public static void main(String[] args) throws JMSException {
    ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
    RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
    // 是否在每次尝试重新发送失败后,增长这个等待时间
    redeliveryPolicy.setUseExponentialBackOff(true);
    // 重发次数,默认为6次 这里设置为1次
    redeliveryPolicy.setMaximumRedeliveries(2);
    // 重发时间间隔,默认为1秒
    redeliveryPolicy.setMaximumRedeliveries(1000);
    // 第一次失败后重新发送之前等待500毫秒,第二次失败再等待500 * 2毫秒,这里的2就是value
    redeliveryPolicy.setBackOffMultiplier(2);
    // 最大传送延迟，只在useExponentialBackOff为true时有效（V5.5），假设首次重连间隔为10ms，倍数为2，那么第二次重连时间间隔为
    // 20ms，
    // 第三次重连时间间隔为40ms，当重连时间间隔大的最大重连时间间隔时，以后每次重连时间间隔都为最大重连时间间隔。
    redeliveryPolicy.setMaximumRedeliveryDelay(1000);
    connectionFactory.setRedeliveryPolicy(redeliveryPolicy);

    Connection connection = connectionFactory.createConnection();
    connection.start();
    Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
    Queue queue = session.createQueue(QUEUE_NAME);
    MessageConsumer consumer = session.createConsumer(queue);

    TextMessage message = (TextMessage) consumer.receive(1000);
    System.out.println("消费者接收到消息:"+message.getText());
    consumer.close();
    session.close();
    connection.close();
    System.out.println("消息消费成功");
}

SpringBoot配置：在配置类中

@Bean
@Primary
public ActiveMQConnectionFactory activeMQConnectionFactory(){
    ActiveMQConnectionFactory acticeMQConnectionFactory = new ActiveMQConnectionFactory(username,password,borkerUrl);
    acticeMQConnectionFactory.setUseAsyncSend(true);//同步还是异步
    RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
    // 是否在每次尝试重新发送失败后,增长这个等待时间
    redeliveryPolicy.setUseExponentialBackOff(true);
    // 重发次数,默认为6次 这里设置为1次
    redeliveryPolicy.setMaximumRedeliveries(2);
    // 重发时间间隔,默认为1秒
    redeliveryPolicy.setMaximumRedeliveries(1000);
    // 第一次失败后重新发送之前等待500毫秒,第二次失败再等待500 * 2毫秒,这里的2就是value
    redeliveryPolicy.setBackOffMultiplier(2);
    // 最大传送延迟，只在useExponentialBackOff为true时有效（V5.5），假设首次重连间隔为10ms，倍数为2，那么第二次重连时间间隔为
    // 20ms，
    // 第三次重连时间间隔为40ms，当重连时间间隔大的最大重连时间间隔时，以后每次重连时间间隔都为最大重连时间间隔。
    redeliveryPolicy.setMaximumRedeliveryDelay(1000);
    connectionFactory.setRedeliveryPolicy(redeliveryPolicy);
}

死信队列

简介

官方介绍：http://activemq.apache.org/message-redelivery-and-dlq-handling.html。

DLQ-死信队列（Dead Letter Queue）用来保存处理失败或者过期的消息。

就是一条消息现次被重发了多次后（默认6次），将会被activemq移入"死信队列"。开发人员可以在这个Queue中查看处理出错的消息，进行人工处理。

开发中的情况

一般生产环境中在使用MQ的时候设计两个队列：一个是核心业务队列，一个是死信队列。

核心业务队列，就是处理正常的消息，另一个就是处理处理异常情况的。

由上图中可以看到，在第三方物流系统故障期间，所有的订单消息全部处理失败，全部会转入死信队列。然后你的仓储系统得专门有一个后台线程。监控第三方物流系统是否正常。能否请求。不停的监视。一旦发现对方恢复正常。这个后台线程就从死信队列消费出来处理失败的订单，重新执行发货和配送的通知逻辑。

配置方式

activemq.xml

<destinationPolicy>
    <policyMap>
        <policyEntries>
            <policyEntry queue=">">
                <deadLetterStrategy>
                    <individualDeadLetterStrategy queuePrefix="DLQ."
                                                  useQueueForQueueMessages="true" />
                </deadLetterStrategy>
            </policyEntry>
            <policyEntry topic=">" >
                <!-- The constantPendingMessageLimitStrategy is used to prevent
                         slow topic consumers to block producers and affect other consumers
                         by limiting the number of messages that are retained
                         For more information, see:

                         http://activemq.apache.org/slow-consumer-handling.html

                    -->
                <pendingMessageLimitStrategy>
                    <constantPendingMessageLimitStrategy limit="1000"/>
                </pendingMessageLimitStrategy>
            </policyEntry>
        </policyEntries>
    </policyMap>
</destinationPolicy>