ActiveMQ_036：哪些消息会进入死信队列？

环境：MAC OS X 10.11.5 + ActiveMQ 5.13.3

1.  超时消息
（1）默认情况下，设置超时的持久化消息超时后，会进入死信队列。
./activemq/bin/activemq producer --brokerUrl tcp://localhost:61616 --user admin --password admin --destination queue://testQueue --persistent true --messageCount 1 --parallelThreads 1 --msgTTL 1000 --messageSize 1
（2）默认情况下，设置超时的非持久化消息超时后，不会进入死信队列，即直接丢弃。
./activemq/bin/activemq producer --brokerUrl tcp://localhost:61616 --user admin --password admin --destination queue://testQueue --persistent false --messageCount 1 --parallelThreads 1 --msgTTL 1000 --messageSize 1

2. 带事务的消费者的 session 调用了 rollback() 方法

import java.io.IOException;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Queue_Receiver_Transaction_Async {

    public static void main(String[] args) {

        // ConnectionFactory：连接工厂，JMS用它创建连接
        ConnectionFactory connectionFactory;
        // Connection：JMS客户端到JMS Provider的连接
        Connection connection = null;
        // Session：一个发送或接收消息的线程
        final Session session;
        // Destination：消息的目的地;消息发送给谁.
        Destination destination;
        // MessageConsumer：消息消费者
        MessageConsumer consumer;

        connectionFactory = new ActiveMQConnectionFactory(
                "admin",
                "admin",
                "failover:(tcp://127.0.0.1:61616)");
        try {
            connection = connectionFactory.createConnection();
            connection.start();
            session = connection.createSession(true, Session.SESSION_TRANSACTED);
            destination = session.createQueue("FirstQueue");
            consumer = session.createConsumer(destination);
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage tm = (TextMessage) message;
                    try {
                        System.out.println("Received message: " + tm.getText());
                        session.rollback();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
          
            try {
                System.in.read();
            } catch (IOException e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection) {
                    connection.close();
                }
            } catch (Throwable ignore) {
            }
        }
    }
}
向 FirstQueue 发送一条消息，然后运行此程序，输出如下：
run:
 INFO | Successfully connected to tcp://127.0.0.1:61616
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1

由此可见，一共重试了 6 次，加上原本的第 1 次，一共 7 次，7 次全部失败后，消息被放入 DLQ。

3. 带事务的消费者的 session 在调用 commit() 方法之前调用了 close() 方法

import java.io.IOException;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Queue_Receiver_Transaction_Async {

    public static void main(String[] args) {

        // ConnectionFactory：连接工厂，JMS用它创建连接
        ConnectionFactory connectionFactory;
        // Connection：JMS客户端到JMS Provider的连接
        Connection connection = null;
        // Session：一个发送或接收消息的线程
        final Session session;
        // Destination：消息的目的地;消息发送给谁.
        Destination destination;
        // MessageConsumer：消息消费者
        MessageConsumer consumer;

        connectionFactory = new ActiveMQConnectionFactory(
                "admin",
                "admin",
                "failover:(tcp://127.0.0.1:61616)");
        try {
            connection = connectionFactory.createConnection();
            connection.start();
            session = connection.createSession(true, Session.SESSION_TRANSACTED);
            destination = session.createQueue("FirstQueue");
            consumer = session.createConsumer(destination);
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage tm = (TextMessage) message;
                    try {
                        System.out.println("Received message: " + tm.getText());
                        session.close();
                        session.commit();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });

            try {
                System.in.read();
            } catch (IOException e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection) {
                    connection.close();
                }
            } catch (Throwable ignore) {
            }
        }
    }
}

向 FirstQueue 发送一条消息，然后运行此程序，输出如下：
run:
 INFO | Successfully connected to tcp://127.0.0.1:61616
Received message: ActiveMq发送的消息1
javax.jms.IllegalStateException: The Session is closed
    at org.apache.activemq.ActiveMQSession.checkClosed(ActiveMQSession.java:769)
    at org.apache.activemq.ActiveMQSession.commit(ActiveMQSession.java:569)
    at com.travelsky.activemq.Queue_Receiver_Transaction_Async$1.onMessage(Queue_Receiver_Transaction_Async.java:61)
    at org.apache.activemq.ActiveMQMessageConsumer.dispatch(ActiveMQMessageConsumer.java:1404)
    at org.apache.activemq.ActiveMQSessionExecutor.dispatch(ActiveMQSessionExecutor.java:131)
    at org.apache.activemq.ActiveMQSessionExecutor.iterate(ActiveMQSessionExecutor.java:202)
    at org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:133)
    at org.apache.activemq.thread.PooledTaskRunner$1.run(PooledTaskRunner.java:48)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:745)

但是并没有重试 6 次，也没有把消息放入 DLQ，和官方文档描述不符。

4. 使用 CLIENT_ACKNOWLEDGE 模式的消费者的 session 调用了 recover() 方法

import java.io.IOException;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;

public class Queue_Receiver_ClientAck_Async {

    public static void main(String[] args) {

        // ConnectionFactory：连接工厂，JMS用它创建连接
        ConnectionFactory connectionFactory;
        // Connection：JMS客户端到JMS Provider的连接
        Connection connection = null;
        // Session：一个发送或接收消息的线程
        final Session session;
        // Destination：消息的目的地;消息发送给谁.
        Destination destination;
        // MessageConsumer：消息消费者
        MessageConsumer consumer;

        connectionFactory = new ActiveMQConnectionFactory(
                "admin",
                "admin",
                "failover:(tcp://127.0.0.1:61616)");

        try {
            connection = connectionFactory.createConnection();
            connection.start();
            session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
            destination = session.createQueue("FirstQueue");
            consumer = session.createConsumer(destination);
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage tm = (TextMessage) message;
                    try {
                        System.out.println("Received message: " + tm.getText());
                        // 如果不加这句，Broker就收不到确认信息，消息将一直保留
                        // 一个有意思的现象是：如果没加这句，相当于没有提交确认消息
                        // 如果这个消费者一直不退出，那么其它消费者也无法接收这些消息。
                        session.recover();
                        message.acknowledge();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });

            try {
                System.in.read();
            } catch (IOException e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (null != connection) {
                    connection.close();
                }
            } catch (Throwable ignore) {
            }
        }
    }
}

向 FirstQueue 发送一条消息，然后运行此程序，输出如下：
run:
 INFO | Successfully connected to tcp://127.0.0.1:61616
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1
Received message: ActiveMq发送的消息1

由此可见，一共重试了 6 次，加上原本的第 1 次，一共 7 次，7 次全部失败后，消息被放入 DLQ。

参考文献：
1. http://activemq.apache.org/message-redelivery-and-dlq-handling.html